Hi, does anyone know how I can get the result of one task to be passed to another task further downstream with the imperative api?

flow = Flow('foo', tasks=[t1,t2,t3, t4])

flow.add_edge(t1,t2, key='x')
flow.add_edge(t2,t3)
flow.add_edge(t3,t4)

# now t4 needs the results of t1 not t2 or t3

Hi @John Faucett, you should be able to use the same style you used for the

t1 →t2

edge using a key. For example,

flow.add_edge(t1, t4, key='y')

Hi @Jeremiah thanks, and that shouldn’t mess up the run order of t1 > t2 > t3 > t4 ?

However, I’d caution you that

add_edge

is the lowest level API for setting dependencies and may skip some checks that higher level APIs perform (for example, handling constant values without wrapping them as tasks). You might opt for

flow.set_dependencies(t4, keyword_tasks={'y':t1})

or even the task’s own method`t4.set_upstream(t1, flow=flow, key='y')` . To be clear your way will work as expected for your example, but using the slightly higher-level APIs might future proof you a little better if more robust checks are introduced.

ah ok perfect. Thanks for the clarification 🙂

You got it!

@Marvin archive “How to pass task results downstream using imperative API”