Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Hi community, I have an imperative api question:
```flow.set_dependencies(
            upstream_tasks=[A],
            task=B,
            keyword_tasks=dict(item=A)
        )```
Do I need to explicitly set a Task as an `upstream_task` when I already have it as a `keyword_task`? I’ve read that one is to set state dependencies and the other is to set data dependencies. Since a data dependency is implicitly a state dependency, having task `B` as keyword_task should be enough, right?

<@U01F2A93VB7> I think there are two options to do that:
1. Initialize the tasks and then pass data dependencies within the Flow constructor
2. Not using Flow constructor and pass data dependencies as keyword_tasks, as in your code snippet.
Example #1:
```from prefect import Task, Flow


class GetData(Task):
    def run(self):
        return 1


class PlusOneTask(Task):
    def run(self, x):
        return x + 1


data = GetData()
plus_one = PlusOneTask()

with Flow("imp") as flow:
    x = data()
    plus_one(x)```
Example #2:
```from prefect import Task, Flow


class GetData(Task):
    def run(self):
        return 1


class PlusOneTask(Task):
    def run(self, x):
        return x + 1


data = GetData()
plus_one = PlusOneTask()

flow = Flow("imp")
flow.set_dependencies(task=plus_one, keyword_tasks=dict(item=data))```
You are right that you don’t need to set state dependencies via upstream and downstream tasks if you pass data between tasks - this way state dependencies are inferred via passing data between tasks.

image.png

btw, I think in the example you shared, you don’t need to pass  upstream_tasks anymore, otherwise this will create two edges between those tasks:

```from prefect import Task, Flow


class GetData(Task):
    def run(self):
        return 1


class PlusOneTask(Task):
    def run(self, x):
        return x + 1


data = GetData()
plus_one = PlusOneTask()

flow = Flow("imp")


if __name__ == "__main__":
    flow.set_dependencies(
        upstream_tasks=[data], task=plus_one, keyword_tasks=dict(item=data)
    )
    flow.visualize()```

<@U02H1A95XDW>
```You are right that you don't need to set state dependencies via upstream and downstream tasks if you pass data between tasks - this way state dependencies are inferred via passing data between tasks```
Great, thank you, that’s what I was wondering about! :ok_hand:

About the examples you’ve given: Example 1 uses the functional api and Example 2 uses the imperative api, my question was only in regards to the imperative api (<https://docs.prefect.io/core/concepts/flows.html#apis>)