<@ULVA73B9P> in prefect v2 I could trigger a subfl...
# ask-marvin
m
@Marvin in prefect v2 I could trigger a subflow flow run with create_flow_run_from_deployment e.g.
Copy code
client = get_client()
flow_run = await client.create_flow_run_from_deployment(
    deployment_id=transfer_deployment_id,
    parameters={
        "param": 'abc',
    },
)
<http://logger.info|logger.info>(f"Created flow run: {flow_run.id}")
The flow that triggered the subflow would wait until the subflow completed. In prefect v3 this works differently, the flow is triggered but not as a subflow. what is the recommended approach in prefect v3?
m
thought for 216.2 seconds
Great question — this changed in Prefect 3.x. Recommended approach in Prefect 3.x: - Use
prefect.deployments.run_deployment(...)
from inside a flow. It replaces
create_flow_run_from_deployment(...)
for this use case and will: - Create the child run - Link it as a subflow in the UI by default (
as_subflow=True
) - Optionally wait for completion (
timeout
) Example (wait for completion and propagate failures):
Copy code
from prefect import flow, get_run_logger
from prefect.deployments import run_deployment

@flow
async def parent_flow():
    logger = get_run_logger()

    child = await run_deployment(
        "transfer/transfer-deployment",   # "flow_name/deployment_name" or deployment ID
        parameters={"param": "abc"},
        timeout=None,                     # wait until the child finishes; use 0 for fire-and-forget
        as_subflow=True,                  # default; shows as a subflow in the UI
    )

    # Optionally propagate child failures to the parent
    if child.state.is_failed():
        exc = await child.state.result(raise_on_failure=False)
        raise exc if isinstance(exc, Exception) else RuntimeError(child.state.message)

    <http://logger.info|logger.info>(f"Created child flow run: {child.id}")
Key notes: - If you trigger via the low-level client (
get_client().create_flow_run_from_deployment(...)
), it will not be linked as a subflow unless you also set a
parent_task_run_id
.
run_deployment()
handles this for you automatically when called inside a flow. - If you want true inline subflow behavior (same process, automatic failure propagation and cancellation), import the flow function and call it directly as
await child_flow(param="abc")
instead of using a deployment. That won’t use a worker or separate infrastructure, though. API reference/source: - run_deployment source