<@ULVA73B9P> In Prefect 3, is there a context mana...
# ask-marvin
n
@Marvin In Prefect 3, is there a context manager I can use to make sure a portion of code is never run in parallel by flows? Basically, I want the
concurrency
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?
m
thought for 461.1 seconds
Short answer: No. In Prefect 3.x the
concurrency(...)
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:
Copy code
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):
Copy code
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):
Copy code
import threading
from contextlib import contextmanager

_local_lock = threading.Lock()

@contextmanager
def local_lock():
    with _local_lock:
        yield
- Same machine, multiple processes (Unix):
Copy code
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:
Copy code
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.