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?

@Fina Silva-Santisteban 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.

@Anna Geller

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! 👌 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)