Balazs
11/11/2025, 12:58 PMMarvin
11/11/2025, 12:58 PMMarvin
11/11/2025, 1:03 PMwith tags(...) context. No code changes inside the task body.
- Scope: All task runs (across flows/deployments/work pools) that carry the tag.
- Multi-constraint composition: If a task has multiple tags with limits, Prefect won’t start it unless all tag limits have a slot available—this is great for modeling composite resource constraints.
- Resource efficiency: Because the task does not enter Running until a slot is available, your worker does not sit blocked—slots are negotiated by the server and the worker is free to pick other work.
- Features: Pure concurrency cap; no rate-decay. Ops-friendly knobs you can change centrally without code changes.
- Docs: Tag-based concurrency limits
- Global concurrency limits
- What they do: Provide a general-purpose “slot” or “token” you acquire in code using context managers; useful anywhere (in a task, a flow, or even external Python code).
- How they’re applied: You explicitly acquire the limit in code, e.g. with concurrency("name"): or with rate_limit("name", slot_decay_per_second=...).
- Scope: Whatever code path you wrap; not tied to tags or orchestration state transitions.
- Flexibility: Can do rate limiting via slot_decay_per_second, partial occupancy, enable/disable, etc.
- Trade-off: If you block while waiting for a slot, that Python process/thread is occupied unless you design around it.
- Docs: Global concurrency limits
Why use tag-based limits at all?
- Guardrails without code changes: Ops can cap “all tasks that hit Postgres” by setting a limit on the postgres tag. Every team that tags appropriately benefits without modifying task code.
- Orchestration-level fairness: Limits are enforced before tasks start, so workers don’t get tied up waiting. Tasks show as AwaitingConcurrencySlot in the UI—clear visibility and no thread blocking.
- Composite constraints, declaratively: Tag a task ["postgres", "heavy"] and enforce that it only runs when both resource pools have capacity. Doing this correctly with multiple global limits requires carefully acquiring multiple names in code.
- Works across everything that uses tags: Multiple flows/deployments sharing the same resource pattern get the same policy, automatically.
When to prefer global concurrency limits instead
- You need rate limiting (token decay) instead of a simple cap.
- You want to throttle only a portion of a task/flow body.
- You need limits in non-Prefect code paths or outside of task orchestration.
- You need programmatic control (e.g., dynamic occupancy or conditional acquisition).
Quick examples
- Tag-based
- Set a cap:
prefect concurrency-limit create postgres 5
- Tag your tasks/flows:
from prefect import task, flow, tags
@task(tags=["postgres"])
def run_query(sql):
...
@flow
def etl():
with tags("postgres", "heavy"):
run_query.submit("select ...")
- Inspect:
prefect concurrency-limit inspect postgres
- CLI reference: prefect concurrency-limit CLI
- Global (in code)
```
from prefect.concurrency import concurrency, rate_limit
def do_work():
with concurrency("my-db-pool"):
# critical section that should be limited
...Marvin
11/11/2025, 1:03 PMNate
11/11/2025, 6:26 PM