Kyle McChesney
09/10/2025, 6:00 PMwait_for to it so that it doesn't start until upstream tasks are complete?Marvin
09/10/2025, 6:00 PMMarvin
09/10/2025, 6:06 PMwait_for=[...] with task futures. The subflow won’t start until those futures are complete.
from prefect import flow, task
@task
def extract():
...
@task
def transform():
...
@flow
def child_flow(payload: str):
...
@flow
def parent_flow():
# kick off upstream tasks
e = extract.submit()
t = transform.submit()
# subflow will not start until e and t are done
subflow_future = child_flow(wait_for=[e, t], payload="ready!")
result = subflow_future.result()
2) Use data dependencies (pass futures as inputs)
- Passing task futures as parameters to the subflow also creates an upstream dependency; Prefect will wait for those futures before starting the subflow.
from prefect import flow, task
@task
def extract():
return "data"
@task
def transform(x: str):
return x.upper()
@flow
def child_flow(x: str, y: str):
# runs only after both x and y are available
...
@flow
def parent_flow():
e = extract.submit()
t = transform.submit(e)
# child_flow will wait on e and t before starting; it receives their results
subflow_future = child_flow(x=e, y=t)
subflow_future.result()
Triggering a deployment as a “subflow”
- If you’re using run_deployment, it does not have a wait_for argument, but you can gate it the same ways:
- Wait on futures before calling it
- Wrap it in a task and use that task’s wait_for
from prefect import flow, task
from prefect.futures import wait
from prefect.deployments import run_deployment
@task
def extract(): ...
@task
def preprocess(): ...
@task
def trigger_child_deployment(params: dict):
return run_deployment("child-flow/production", parameters=params)
@flow
def parent_flow():
e = extract.submit()
p = preprocess.submit()
# Option A: wrap in a task and use wait_for
dep = trigger_child_deployment.submit({"k": "v"}, wait_for=[e, p])
dep.result()
# Option B: explicitly wait, then call inline
wait([e, p])
run_deployment("child-flow/production", parameters={"k": "v"})
Notes:
- Tasks have a wait_for on .submit(...).
- Subflows (flow calls) have wait_for on the flow call itself.
- Passing futures as parameters also creates dependencies automatically.
If you can share whether you’re calling a code-defined subflow or triggering a deployment, I can tailor the example exactly to your setup.Kyle McChesney
09/10/2025, 11:21 PMMarvin
09/10/2025, 11:21 PMMarvin
09/10/2025, 11:23 PMrun_deployment, you’re launching a separate flow run (a deployment) and you don’t get a PrefectFuture directly. The best way to get the child flow’s return value is:
- Create the flow run with run_deployment(...)
- Wait for it to finish with wait_for_flow_run(...)
- Extract the returned value from its terminal state with get_state_result(...)
Here are two solid patterns:
1) Inline (blocks the parent until the child completes)
from prefect import flow
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
from prefect.states import get_state_result
@flow
def parent_flow():
# Start the child deployment
fr = run_deployment(
"child-flow/production",
parameters={"k": "v"},
as_subflow=True, # keeps lineage in the UI
)
# Wait for completion
fr = wait_for_flow_run(fr.id, timeout=3600)
# Get the child flow's return value (raises on failure by default)
child_result = get_state_result(fr.state)
return child_result
2) Wrap in a task to get a Future in the parent DAG
- This lets you combine with wait_for=[...] or other task futures.
from prefect import flow, task
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
from prefect.states import get_state_result
@task
def trigger_and_get_child(params: dict):
fr = run_deployment("child-flow/production", parameters=params, as_subflow=True)
fr = wait_for_flow_run(fr.id, timeout=3600)
return get_state_result(fr.state)
@flow
def parent_flow():
# ... upstream work here, producing futures u1, u2, etc.
fut = trigger_and_get_child.submit({"k": "v"}) # or pass wait_for=[u1, u2]
child_result = fut.result()
return child_result
Notes and gotchas:
- Failures: get_state_result(fr.state) will raise if the child flow failed/crashed; catch exceptions if you want to handle them.
- Serialization: The child flow’s return value must be serializable by your configured result serializer/storage. Simple Python types are fine. If you return large/complex objects, ensure result storage is configured and the serializer can handle it.
- Lineage: Use as_subflow=True (default) so the run appears as a subflow of the parent in the UI.Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.
Powered by