DJ Erraballi

11/09/2020, 7:21 PM
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.
Copy code
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
    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=????)

Kyle Moon-Wright

11/09/2020, 7:47 PM
Hey @DJ Erraballi, Using the imperative API, we can set data dependencies using the
kwarg and pass the data between tasks. Here's an example of using the kwarg in which the arg for my
is set to the result of my
Copy code
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})

DJ Erraballi

11/09/2020, 7:48 PM
👍 awesome
can i reference a function in the imperative api? Or is that mixing syntax?

Kyle Moon-Wright

11/09/2020, 7:54 PM
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.

DJ Erraballi

11/10/2020, 6:18 PM
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