<@ULVA73B9P> Do you have example code somewhere in...
# ask-marvin
m
@Marvin Do you have example code somewhere in docs for how to run a sub-flow from within a flow, asynchronously?
m
thought for 192.9 seconds
Yes — there are a couple of documented patterns in Prefect 3.x for running a subflow “asynchronously,” depending on whether you want concurrent nested subflows (parent still waits at some point) or true fire‑and‑forget (parent does not wait). - Concurrent nested subflows (await concurrently with asyncio.gather) - Docs: Run work concurrently and Flows concept - Example:
Copy code
import 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:
Copy code
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.