https://prefect.io logo
b

Brennan Tolman

08/16/2023, 7:12 PM
@Marvin I have two separate flows where the second is dependent on the first and also needs to use the results of the first. How can I do that in Prefect?
m

Marvin

08/16/2023, 7:13 PM
Ah, the joy of dependencies in Prefect. It's a bit like trying to untangle a ball of yarn that a kitten has been playing with, isn't it? In Prefect, you can use the
set_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:
Copy code
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?
n

Nate

08/16/2023, 7:15 PM
gah @Marvin - that's not right. you can do
Copy code
from 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?
m

Marvin

08/16/2023, 7:16 PM
Oh, I see. You've decided to take the high road and use Prefect's decorator syntax to define your flows. That's certainly one way to do it, and it's a bit cleaner than the method I suggested. In your example, you're defining two flows:
first
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.
b

Brennan Tolman

08/16/2023, 7:19 PM
@Nate Kind of, but I was looking to put the flows in different scripts and was mostly wondering if there was a way to point to the status and results of another flow through the api. Does that make sense?
n

Nate

08/16/2023, 7:21 PM
yeah, so in the example above you'd be free to define the flows elsewhere and import / call the python object alternatively if you want them decoupled from an infra perspective you could make each a deployment and then call each with
run_deployment
like
Copy code
@flow
def parent():
   foo: FlowRun = run_deployment("foo/deployment")
   run_deployment("needs-foo/deployment", parameters=dict(foo=foo.state.result()))