@Marvin How can I await for the state of a subflow deployed using run_deployment to be done

To await the completion of a subflow deployed using

run_deployment

in Prefect 3.x, you can use the

wait_for_flow_run

function. This function allows you to monitor and wait for the state of a flow run, including subflows, until it reaches a terminal state (e.g., completed, failed, etc.). Here's a basic example of how you might use it:

python
from prefect import flow
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run

@flow
def parent_flow():
    # Trigger the subflow deployment
    deployment_id = run_deployment(name="my_subflow_deployment")
    
    # Wait for the subflow to complete
    state = wait_for_flow_run(deployment_id)
    
    # Check the state of the completed subflow
    if state.is_completed():
        print("Subflow completed successfully.")
    else:
        print(f"Subflow ended with state: {state}")

In this example: -

run_deployment

starts the subflow. -

wait_for_flow_run

waits for the subflow to reach a terminal state. - You can then check the state to determine the outcome. For more detailed information, you can refer to the Prefect 3.x documentation on deployments.

@Marvin can we pass multiple FlowRuns ?

To wait for multiple subflows to complete in Prefect 3.x, you can use the

as_completed

function from the

prefect.futures

module. This allows you to handle multiple flow runs concurrently and process them as they complete. Here's a conceptual example:

python
from prefect import flow
from prefect.deployments import run_deployment
from prefect.futures import as_completed
from prefect.flow_runs import wait_for_flow_run

@flow
def parent_flow():
    # Trigger multiple subflow deployments
    deployment_ids = [run_deployment(name=f"subflow_deployment_{i}") for i in range(3)]
    
    # Wait for all subflows to complete
    for future in as_completed([wait_for_flow_run(deployment_id) for deployment_id in deployment_ids]):
        state = future.result()
        if state.is_completed():
            print(f"Subflow {state.run_id} completed successfully.")
        else:
            print(f"Subflow {state.run_id} ended with state: {state}")

In this example: -

run_deployment

triggers multiple subflows. -

as_completed

is used to process each subflow as it completes. -

wait_for_flow_run

is called for each subflow to wait for its terminal state. This pattern allows you to handle multiple subflows efficiently and react to each one's completion as it happens. For further details, you might want to explore the Prefect 3.x documentation.