Etienne Callies
03/09/2026, 1:21 PMMarvin
03/09/2026, 1:21 PMMarvin
03/09/2026, 1:27 PMprefect.concurrency.concurrency context manager) — they are not configurable for deployment-level concurrency or tag-based task concurrency. Here’s how each case works and what you can configure instead:
- Deployment-level concurrency
- There is no lease duration or strict mode here.
- You control:
- `concurrency_limit`: maximum concurrent runs for that deployment
- `concurrency_options.collision_strategy`: what to do when the limit is hit (ENQUEUE or CANCEL_NEW)
- `concurrency_options.grace_period_seconds`: buffer for infra startup before a run “consumes” a slot (not a lease)
- Example (Python deploy):
from prefect import flow
from prefect.server.schemas.core import ConcurrencyOptions, ConcurrencyLimitStrategy
@flow
def my_flow():
...
my_flow.from_source(
source="<https://github.com/org/repo.git>",
entrypoint="flows.py:my_flow",
).deploy(
name="my-deployment",
work_pool_name="my-pool",
concurrency_limit=5, # simple cap on concurrent runs
concurrency_options=ConcurrencyOptions(
collision_strategy=ConcurrencyLimitStrategy.ENQUEUE, # or CANCEL_NEW
grace_period_seconds=300,
),
)
- Tag-based task concurrency
- There is no user-facing lease duration or strict mode.
- You set a limit per tag, and Prefect automatically delays tasks when the limit is reached. The server returns a Retry-After and the client backs off.
- The main “knob” is the server-side cap on the wait time for tag-based slots:
- PREFECT_SERVER_TASKS_TAG_CONCURRENCY_SLOT_WAIT_SECONDS (default 30s)
- Usage:
- Tag tasks:
from prefect import task
@task(tags=["db"])
def query(...):
...
- Manage limits via CLI (verified):
prefect concurrency-limit create db 10
prefect concurrency-limit inspect db
prefect concurrency-limit ls
prefect concurrency-limit reset db
prefect concurrency-limit delete db
- Or via Python client:
from prefect import get_client
async with get_client() as client:
await client.create_concurrency_limit(tag="db", concurrency_limit=10)
limit = await client.read_concurrency_limit_by_tag(tag="db")
await client.reset_concurrency_limit_by_tag(tag="db")
await client.delete_concurrency_limit_by_tag(tag="db")
If you specifically need to set lease duration and strict mode, use global concurrency around the code that should be gated. That’s where these options live:
- Global concurrency (for completeness)
- Available knobs on the context manager:
- lease_duration (seconds; default 300)
- strict (default False)
- Signature (verified):
from prefect.concurrency.sync import concurrency
# Parameters:
# names, occupy=1, timeout_seconds=None, max_retries=None,
# lease_duration=300, strict=False, holder=None
with concurrency("resource-name", occupy=1, lease_duration=600, strict=True):
# protected section
...
Want help mapping these to your specific flow(s) and deployment(s)? If you share how you’re currently setting your deployment or tags, I can suggest the exact changes.