hey all? what's the best way to create flow dependencies in prefect 2.0? We used a state handler to do it in prefect 1.0 and we're looking to migrate to 2.0 but scoping out a few different things first. our particular use case is we have a few different flows that extract data from sources and push the data to our data lake, then our redshift cluster. something like elt_1 elt_2 elt_3 once all the different elts are done, we want to trigger our higher level transform flow that preps data for looker. we want to wait till all 3 elts are done and not trigger the final transform until all parents are completed

I found this. Try using

wait_for

in the subflow call. https://github.com/PrefectHQ/prefect/pull/7343

There’s an example in the github of recipes

yep, to give an example of what Dominic is describing

from typing import Any
from prefect import flow, task

@flow
def elt(input_arg: Any):
   print(f"got {input_arg!r}, executing elt")
   return True

@task
def final_transform():
   print("doing final transform!")

@flow(log_prints=True)
def orchestrator():
   elt_results = [elt(elt_input) for elt_input in ["input1", "input2", "input3"]]
     
   final_transform(wait_for=elt_results)

if __name__ == "__main__":
   orchestrator()

or if

elt

is a task, you could

map

it to simplify the syntax, or if its a standalone deployment you could use

run_deployment

gotcha, so we would have to make a flow of flows for this? there's no way to have totally distinct flows that trigger one another?

the above is just one way, you can definitely trigger flows from one another for example, if you have a deployed flow already, you can call

run_deployment

from another flow (or anywhere else you're authed)

say elt1 and elt2 finish at the same time, could I block one from running and timeout within a few seconds if it doesn't start?