@Marvin is there a way when submitting a new flow run, make it wait until another flow run is finished? just like task has an argument wait_for to wait for other tasks

In Prefect 3.x, there isn't a direct built-in feature to make a flow run wait for another flow run to complete, similar to how tasks can wait for other tasks using

wait_for

. However, you can achieve this behavior using a combination of flow state management and orchestration. Here's a general approach you can follow: 1. Use Flow States: You can check the state of a flow run using the Prefect client. You can implement a polling mechanism to wait until the previous flow run reaches a terminal state (such as

Completed

). 2. Pause and Resume: Utilize the

pause_flow_run

and

resume_flow_run

functionalities to manage flow execution. This involves pausing a flow until the prerequisite flow run is completed. Here's a simple example of how you might structure this logic:

python
from prefect import flow, get_client, State

@flow
def dependent_flow():
    client = get_client()
    # Assuming you have the flow run ID of the flow you want to wait for
    flow_run_id_to_wait_for = "your-flow-run-id"

    # Poll the state of the flow run
    while True:
        state = client.read_flow_run(flow_run_id_to_wait_for).state
        if state.is_completed():
            break

    # Proceed with the rest of your flow logic
    print("The prerequisite flow run has completed.")

# Example usage
dependent_flow()

For more detailed implementations, you might want to explore Prefect's orchestration capabilities further. You can refer to the Prefect API documentation for more details on managing flow runs and their states.