Short answer: you can get “one-at-a-time per id” behavior, but Prefect’s global concurrency limits are name-scoped. There isn’t a single parameterized limit that enforces “1 per arbitrary key” without creating a record per key. The usual pattern is to use dynamic names (e.g.,

customer:123

) and create those limits lazily so you don’t have to pre-enumerate all keys. How to do it - Use the concurrency context with a dynamic name derived from your id. This ensures only one run can hold that name at a time, while different ids can proceed concurrently:

from prefect.concurrency.sync import concurrency

def do_work(id: int):
    resource = f"my-op:{id}"
    # blocks here until a slot for this resource is available
    with concurrency(names=resource, strict=True, timeout_seconds=600):
        # exactly one flow run can be in this block per id
        ... your operation ...

- You’ll need a global concurrency limit record per name (set to 1). Prefect does not auto-create a limit for a name you reference. To avoid pre-creating hundreds of limits, create them lazily on first use with the Prefect client (or run a small bootstrap flow that seeds limits for the active ids you care about). After that, the

concurrency(...)

call will enforce the lock. Notes -

strict=True

makes acquisition fail/block if the named limit does not exist; use this once you’re creating the limit programmatically or via UI. - You can compose limits if needed:

names=[f"my-op:{id}", "shared-db"]

to acquire multiple locks at once. - There’s no CLI subcommand to manage concurrency limits in Prefect 3.x; use the UI or the Python client. If you’re okay with lazy creation, I can share a tiny helper that: - checks for a global concurrency limit with name

my-op:{id}

- creates it with limit=1 if missing - then enters

concurrency(names=..., strict=True)

Let me know if you’re on Prefect Cloud or self-hosted, and I’ll tailor the snippet.