@Marvin Can i limit the concurrent executions for a specific workflow run?

Yes, you can limit the concurrent executions for a specific workflow run in Prefect. This can be achieved by using the

concurrency

context manager. For example, if you want to limit the number of

process_data

tasks to five at any one time, you can request five slots on the

data-processing

concurrency limit. Here is an example of how you can use the

concurrency

context manager:

with concurrency("database", occupy=1):
    connection = psycopg2.connect("<connection_string>")
    cursor = connection.cursor()
    cursor.execute(query)
    result = cursor.fetchall()
    connection.close()
    return result

@flow
def my_flow():
    queries = ["SELECT * FROM table1", "SELECT * FROM table2", "SELECT * FROM table3"]
    for query in queries:
        database_query.submit(query)

if __name__ == "__main__":
    my_flow()

In this example, the

concurrency

context manager is used to ensure that we never exceed the maximum number of database connections. For more details, you can refer to the Prefect documentation.

@Marvin does it work between flow runs?

The concurrency context manager in Prefect is designed to limit the number of concurrent tasks within a single flow run. It does not limit concurrency across multiple flow runs. If you want to limit the concurrency across multiple flow runs, you would need to manage this at the level of your Prefect infrastructure (e.g., by limiting the number of workers available to execute tasks). For more complex scenarios, you might consider using Prefect's server or cloud offerings, which provide more advanced orchestration capabilities.