Trying to wrap my head around the imperative api, I feel like I want a little more fine grained control on the graph than what the functional API provides, but its really hard for me to understand how i can pass data between tasks. Ex.

Task A: 
  run(self, x, y):

Task B: 
  run(self, a_result, x, y):


with Flow() as flow: 
    x_parameter = Parameter("x", required=True)
    y_parameter = Parameter("y",  required=True)
    # question is how do setup the parameter pasing from teh parameters -> A -> B
    # my attempt but still dont know how to keyword_tasks in this scenario to pass along results from A
    flow.add_task(x_parameter) 
    flow.add_task(y_parameter)
    a = A() 
    flow.set_dependencies(a , upstream_tasks=[x_parameter, y_parameter], keyword_tasks=??? )
    flow.set_dependencies(B(), upstream_tasks=[a, x_parameter, y_parameter], keyword_tasks=????)

Hey @DJ Erraballi, Using the imperative API, we can set data dependencies using the

keyword_tasks

kwarg and pass the data between tasks. Here's an example of using the kwarg in which the arg for my

TaskB

is set to the result of my

TaskA

instance:

from random import randrange
from prefect import Task, Flow

class TaskA(Task):
    def run(self):
        return randrange(1, 100)

class TaskB(Task):
    def run(self, data):
        return data * 10

a = TaskA()
b = TaskB()

flow = Flow("Example Flow")

flow.set_dependencies(b, keyword_tasks={'data': a})

flow.run()

👍 awesome

At the end of the day, your flow needs to consist of tasks - so you can certainly use the task decorator on a function and incorporate that into your flow with the imperative API. If you are referring to generic functions - your tasks can certainly reference other functions, but don't expect them to show in the dependency graph if they're not defined as tasks.

Just wondering how if in the example above if taskA was defined functionaly, we would be able to pass around the return values in keyword_tasks, or set upstreams with a function rather than an instance of a class