@Marvin How can i use the

concurrency

context manager from

prefect.concurrency.sync

to limit the number of subflows running at a time in my main flow?

You can use the

concurrency

context manager from

prefect.concurrency.sync

to limit the number of simultaneous executions of a given portion of your code. It will not create a subflow itself but can be used within tasks to control the execution of certain parts of your flow. Here is an example from the Prefect Docs:

python
from prefect import flow, task
from prefect.concurrency.sync import concurrency

@task
def process_data(x, y):
    with concurrency("database", occupy=1):
        return x + y

@flow
def my_flow():
    for x, y in [(1, 2), (3, 4), (3, 4), (4, 5)]:
        process_data.submit(x, y)

if __name__ == "__main__":
    my_flow()

The above code does not limit subflows, but you can adapt the concept for your subflows. When you have a task that initiates subflows, you can place that task inside the

with concurrency

context manager and use an appropriate concurrency limit name and resource limit. This will ensure that the number of concurrent subflows being executed will not exceed the specified limit. Remember that subflows will be treated as separate flow runs once initiated, as described in the Flows Concept Documentation. So the concurrency context is for controlling simultaneous task executions rather than the subflows themselves. If you want to limit the concurrency of subflows specifically, you'll probably need to design your main flow to control when to start subflows based on some condition or state, because the

concurrency

context manager is not designed to limit subflow runs.