Martin Janiczek
12/01/2025, 12:06 PMMarvin
12/01/2025, 12:06 PMMarvin
12/01/2025, 12:09 PMimport asyncio
from prefect import flow
@flow
async def child_flow(x: int) -> int:
await asyncio.sleep(1)
return x * 2
@flow
async def parent_flow():
# Run child subflows concurrently
results = await asyncio.gather(
child_flow(1),
child_flow(2),
child_flow(3),
)
return results
Notes:
- These are nested flow runs; you’re running them concurrently, but you still await them (e.g., via asyncio.gather).
- If you need subflows that can outlive the parent or be canceled independently, use deployments (below).
- Fire‑and‑forget subflows via deployment triggers (don’t wait)
- Docs: Trigger ad‑hoc deployment runs
- API: run_deployment reference
- Example:
from prefect import flow
from prefect.deployments import run_deployment
@flow
def parent_flow():
# Start deployed flow runs and return immediately
fr1 = run_deployment("my-child-flow/my-deployment", parameters={"x": 1}, timeout=0)
fr2 = run_deployment("my-child-flow/my-deployment", parameters={"x": 2}, timeout=0)
fr3 = run_deployment("my-child-flow/my-deployment", parameters={"x": 3}, timeout=0)
return [fr1, fr2, fr3] # FlowRun objects you can track later
Notes:
- timeout=0 returns immediately without waiting.
- These runs are linked as subflows by default (as_subflow=True) and can be managed independently of the parent.
Quick guidance:
- If you just need concurrency inside one flow run and you’ll await results: use asyncio.gather with nested subflows.
- If you truly don’t want to wait (and want independent lifecycle/cancellation): use run_deployment(..., timeout=0).
Is your goal concurrent nested subflows that you’ll eventually await, or fire‑and‑forget child runs that outlive the parent? I can share a more tailored example once I know which you prefer.