Brennan Tolman
08/16/2023, 7:12 PMMarvin
08/16/2023, 7:13 PMset_upstream
and set_downstream
methods to define dependencies between tasks. But you're asking about flows, not tasks. Unfortunately, Prefect doesn't support dependencies between flows out of the box.
However, you could create a "master" flow that triggers the two dependent flows as tasks using the FlowRunTask
. The first flow's results can be passed to the second flow as parameters. Here's a rough sketch of how you might do it:
python
from prefect import Flow
from prefect.tasks.prefect import FlowRunTask
flow_run_task = FlowRunTask()
with Flow("Master Flow") as master_flow:
flow1_result = flow_run_task(flow_name="Flow 1")
flow2_result = flow_run_task(flow_name="Flow 2", parameters={"input": flow1_result})
master_flow.run()
In this example, "Flow 1" and "Flow 2" would be the names of your two flows. The results of "Flow 1" are passed to "Flow 2" as a parameter.
Remember, though, that this is a workaround. It's not as elegant as it could be, and it might not work for all use cases. But then, what in life is?Nate
08/16/2023, 7:15 PMfrom prefect import flow
@flow
def first():
return "foo"
@flow
def needs_foo(foo: str):
print(foo)
@flow
def parent():
foo = first()
needs_foo(foo)
are you looking for something else?Marvin
08/16/2023, 7:16 PMfirst
and needs_foo
. The first
flow returns a string, and the needs_foo
flow takes that string as an argument. Then, in the parent
flow, you're calling both of these flows in sequence.
This is a perfectly valid way to set up dependencies between flows in Prefect. The parent
flow acts as a sort of orchestrator, calling the other flows in the correct order and passing data between them.
I must say, it's quite refreshing to see someone using Prefect's advanced features like this. It's almost enough to make me feel optimistic. Almost.Brennan Tolman
08/16/2023, 7:19 PMNate
08/16/2023, 7:21 PMrun_deployment
like
@flow
def parent():
foo: FlowRun = run_deployment("foo/deployment")
run_deployment("needs-foo/deployment", parameters=dict(foo=foo.state.result()))