How to limit concurrent task runs with tags

from prefect import flow, task

@task(tags=["database"])
def fetch_data():
    # This task will respect the "database" tag's concurrency limit
    return "data"

@flow
def my_workflow():
    fetch_data()

from prefect import flow, task

@task(tags=["database"])
def query_database(query: str):
    # Simulate database work
    return f"Results for {query}"

@flow
def data_pipeline():
    # These will respect the "database" tag limit
    query_database("SELECT * FROM users")
    query_database("SELECT * FROM orders")
    query_database("SELECT * FROM products")

from prefect import task

@task(tags=["database", "analytics"])
def complex_query():
    # This task needs available slots in both "database" AND "analytics" limits
    return "complex results"

# Set a limit of 10 for the "database" tag
prefect concurrency-limit create database 10

# View all concurrency limits
prefect concurrency-limit ls

# View details about a specific tag's limit
prefect concurrency-limit inspect database

# Delete a concurrency limit
prefect concurrency-limit delete database

import asyncio

from prefect import get_client


async def manage_concurrency_limits():
    async with get_client() as client:
        # Set a concurrency limit of 10 on the "database" tag
        await client.create_concurrency_limit(
            tag="database",
            concurrency_limit=10
        )

        # Read current limit for a tag
        limit = await client.read_concurrency_limit_by_tag(tag="database")
        print(limit)

        # View all concurrency limits
        limits = await client.read_concurrency_limits(limit=10, offset=0)
        print(limits)

        # Delete a concurrency limit
        await client.delete_concurrency_limit_by_tag(tag="database")


asyncio.run(manage_concurrency_limits())

# Create a concurrency limit
curl -X POST "http://localhost:4200/api/concurrency_limits/" \
  -H "Content-Type: application/json" \
  -d '{"tag": "database", "concurrency_limit": 10}'

# Get all concurrency limits
curl "http://localhost:4200/api/concurrency_limits/"

resource "prefect_concurrency_limit" "database_limit" {
  tag               = "database"
  concurrency_limit = 10
}

prefect config set PREFECT_SERVER_TASKS_TAG_CONCURRENCY_SLOT_WAIT_SECONDS=60