Thread
#prefect-community
    Fina Silva-Santisteban

    Fina Silva-Santisteban

    10 months ago
    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?
    Anna Geller

    Anna Geller

    10 months ago
    @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.
    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()
    Fina Silva-Santisteban

    Fina Silva-Santisteban

    10 months ago
    @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)