Nicholas Pini
03/11/2026, 9:44 AMconcurrency context manager, but as far as I know it requires creating a concurrency limit in the Prefect server, which is a bit overkill because it is something I need only for a single portion of code. Can I simply use concurrency (or something like it) without needing to setup anything on the server?Marvin
03/11/2026, 9:45 AMMarvin
03/11/2026, 9:52 AMconcurrency(...) context manager always talks to the Prefect API and requires a global concurrency limit to exist. There isn’t a built‑in “local-only” concurrency context manager you can use without creating something on the server.
Details and options:
- Using concurrency(...) without a server-side limit won’t enforce anything. By default it will warn and continue; with strict=True it will error if the limit is missing.
- If you need cross-flow, cross-worker serialization, you’ll need either:
- a Prefect Global Concurrency Limit (GCL), or
- your own external distributed lock (e.g., Redis, DynamoDB, etc.)
Practical patterns
1) Small/just-in-time setup with Prefect’s GCL (recommended for distributed runs)
- You can programmatically “ensure” the limit exists at runtime instead of a manual pre-step:
from prefect.client.orchestration import get_client
from prefect.concurrency.sync import concurrency
LIMIT_NAME = "my-critical-section"
def ensure_limit():
# Idempotent: creates or updates if needed
with get_client(sync_client=True) as client:
client.upsert_global_concurrency_limit_by_name(
name=LIMIT_NAME, limit=1
)
def do_thing():
ensure_limit()
with concurrency(LIMIT_NAME, occupy=1, strict=True):
# This section will never run in parallel across flows/workers
...
- CLI alternative (one-time):
prefect gcl create my-critical-section --limit 1
- Docs:
- Concepts: Global Concurrency Limits
- How-to: Apply Global Concurrency and Rate Limits
- API: prefect.concurrency.sync
2) Local-only locking (works only within a single process or single machine)
- Single process (threads only):
import threading
from contextlib import contextmanager
_local_lock = threading.Lock()
@contextmanager
def local_lock():
with _local_lock:
yield
- Same machine, multiple processes (Unix):
import fcntl
from contextlib import contextmanager
@contextmanager
def file_lock(path="/tmp/my-critical-section.lock"):
with open(path, "w") as f:
fcntl.flock(f, fcntl.LOCK_EX)
try:
yield
finally:
fcntl.flock(f, fcntl.LOCK_UN)
Caveat: these won’t coordinate across separate machines/containers. In AWS CodeBuild or multiple workers, they’re usually not sufficient.
3) External distributed lock (no Prefect server dependency)
- Use Redis (SETNX + TTL), DynamoDB conditional writes, etc., to implement a lock around your critical section. This is a good fit if you truly want no Prefect-side setup but need cross-worker serialization.
A couple of gotchas
- If you try concurrency("name") without creating the limit, it won’t actually protect the section unless you pass strict=True (which will then raise).
- If you only need to prevent parallelism within a single flow run, you can use a sequential task runner for that flow, but that does not serialize across different flows:
from prefect import flow
from prefect.task_runners import SequentialTaskRunner
@flow(task_runner=SequentialTaskRunner())
def my_flow():
...
If you can share how your flows are scheduled/executed (single worker vs multiple, single host vs multiple hosts/containers in CodeBuild), I can recommend the simplest locking approach that will actually hold in your environment.