@Marvin Trigger multiple deployments, A and B at the same time, and C should wait for A (and run regardless if A succeeds or not)

The provided example demonstrates the use of asynchronous flows in Prefect but does not directly address how to trigger multiple deployments with dependencies. In Prefect, to accomplish the behavior you're looking for with deployments A, B and C, you'll need to structure your flow runs and their deployments such that: - Deployments A and B are triggered simultaneously without waiting for each other. - Deployment C is triggered after A has started but does not depend on the success of A. In Prefect 2.x, you might orchestrate this through a combination of

asyncio

and deployment trigger configurations. I suggest using

async

to start A and B concurrently, and then simply starting C after initiating A without the need to check if A was successful. Here's a pseudo-code example to illustrate:

python
import asyncio
from prefect import flow

@flow(name="Deployment A")
def deployment_a():
    # Your logic here

@flow(name="Deployment B")
def deployment_b():
    # Your logic here

@flow(name="Deployment C")
def deployment_c():
    # Your logic here

@flow(name="Orchestrator")
async def orchestrator():
    a_task = asyncio.create_task(deployment_a())  # Start A
    await asyncio.create_task(deployment_b())     # Start B concurrently
    await a_task  # Optionally wait for A to start or finish
    deployment_c()  # Start C

if __name__ == "__main__":
    asyncio.run(orchestrator())

You would set up your deployments to match this logic, using Prefect's deployment CLI or Python API. Note that Prefect's flow execution model may be subject to change, and you should consult the most recent Prefect documentation or community for updates or best practices. For further details, I recommend checking Prefect's documentation regarding deployments, specifically sections on orchestration and dependencies: - Prefect Deployments - Prefect Task Orchestration

@Marvin I want to do the same but a loop over flows 1....n and then afterwards run flow C

The provided example is for a basic flow and does not directly address looping over multiple flows followed by running another flow. However, we can still create a concept for executing multiple flows in a loop and then executing another flow using asyncio for concurrency and Prefect. Below is a conceptual pseudo-code that demonstrates this approach:

python
import asyncio
from prefect import flow

@flow(name="Flow Template")
def flow_template(flow_number):
    # Your logic for each flow here
    print(f"Running Flow #{flow_number}")

@flow(name="Follow-Up Flow")
def follow_up_flow():
    # Your logic for the follow-up flow here
    print("Running Follow-Up Flow")

@flow(name="Orchestrator with loop")
async def orchestrator_with_loop():
    # Start flows 1 to n concurrently
    flow_tasks = [asyncio.create_task(flow_template(i)) for i in range(1, n + 1)]
    # Wait for all flows to be done
    await asyncio.gather(*flow_tasks)
    
    # Run follow-up flow C
    follow_up_flow()

if __name__ == "__main__":
    n = 10  # For example, if you have 10 flows to run
    asyncio.run(orchestrator_with_loop())

This script defines a template flow that can be parameterized by a number. In the orchestrator flow, it loops from 1 to n, starting each flow concurrently, and then, once all n flows have been kicked off, it waits for them to finish. Once all loops are complete, the follow-up flow is executed. Please customize the template for your flows and the follow-up flow logic according to your application requirements. For more tailored examples and best practices on looping over flows or orchestrating complex workflows in Prefect, you would typically search the official Prefect documentation or community discussions for guidance: - Prefect Docs - Prefect Community Slack Recursive or dynamic workflow patterns can be more complex, so it's crucial to consider potential issues like error handling, timeouts, and workflow state management when designing your flows.