coalescer

Collapse concurrent duplicate async calls into a single execution. Hang one decorator on an async function you already have: while a call is in flight, everyone asking for the same thing joins it instead of starting their own.

from coalescer import coalesce


@coalesce()
async def get_article(article_id: int) -> dict:
    return await load_article_from_db(article_id)

That one decorator is the whole difference between these two runs — the SELECT lines the example prints are trimmed here:

$ python examples/articles.py

without coalescer: 100 concurrent readers of article 1
  -> database queries: 100

with @coalesce(): 100 concurrent readers of article 1
  -> database queries: 1

All hundred callers get that one result. Different arguments stay separate, so get_article(2) still runs its own call. No dependencies, fully typed, Python 3.10+.

Install

pip install coalescer

The problem

Duplicate concurrent work is pure waste. N callers want the same thing at the same moment, none of them has it yet, so all N go and do the same expensive thing. Traffic makes it worse instead of better: the busier you are, the more callers fit inside the window where that work is in progress.

nothing is there yet
   │
   ├── caller 1 ──► do the work ──┐
   ├── caller 2 ──► do the work   │  100 identical calls
   ├── caller 3 ──► do the work   │  for one result
   └── ...        ──► do the work ─┘

Where this shows up:

• An access token expires. Every in-flight request notices at the same moment and posts to the token endpoint. One refresh was needed.
• A rate-limited upstream. Identical concurrent calls each spend quota to fetch the same answer.
• An expensive resource on first use. Warming a pool, loading a model, reading remote config — concurrent callers should wait for the first one rather than each build their own.
• A cache entry expires. Every request misses at once and piles into the database until the first one writes the value back — a cache stampede, also called a thundering herd or dogpile. A cache cannot fix this by itself; it can only serve what has already been written.

In all four the missing rule is the same: while one call is in progress, everybody else waits for it instead of starting their own. That is request coalescing, and it is all this library does.

Pairing with a cache

coalescer replaces nothing in your stack — it guards the gap a cache cannot cover. The cache answers hits; @coalesce keeps the misses from multiplying. Any decorator that caches a coroutine composes the same way:

from coalescer import coalesce


@your_cache(...)  # whichever cache decorator you use
@coalesce()
async def get_article(article_id: int) -> dict:
    return await load_article_from_db(article_id)

Both orders work, and they are not the same:

Put the cache on the outside when a cache hit must stay as cheap as possible — the common choice. Put @coalesce on the outside when the cache itself is remote and you would rather not send it a hundred identical GETs.

Key builders stay compatible: coalescer calls key_builder(func, *args, **kwargs), the convention cache decorators use, so one callable can be handed to both and they will agree on the key.

Without the decorator

The decorator is a thin layer over a group you can drive yourself:

from coalescer import Coalescer

group = Coalescer()


async def get_article(article_id: int) -> dict:
    return await group.do(f"article:{article_id}", load_article_from_db, article_id)

Coalescer has three methods worth knowing:

• await group.do(key, fn, *args, **kwargs) — run fn, or join the call already running for key.
• group.forget(key) — stop new callers from joining the current flight. The one in progress still serves its own waiters; the next call starts fresh.
• len(group) — how many flights are in the air.

A decorated function carries its group as an attribute, so you can reach the same controls:

get_article.coalescer.forget("article:1")

Type checkers do not see that attribute — the decorator returns a plain callable so that decorated methods keep binding self correctly. When you want a typed handle, build the group yourself and hand it over:

group = Coalescer()


@coalesce(group=group)
async def get_article(article_id: int) -> dict: ...


group.forget(...)

Options

@coalesce(
    key=None,           # fixed key for every call, ignoring arguments
    key_builder=None,   # key_builder(func, *args, **kwargs) -> Hashable
    noself=False,       # leave `self` out of the key, so instances share flights
    group=None,         # share one Coalescer across several functions
    timeout=None,       # seconds this caller waits before giving up
)

By default the key is a tuple of the function's module, qualified name, and the arguments themselves, so arguments are compared by value — two equal objects share a flight even when their repr() would differ. Arguments that cannot be hashed (a list, a dict, a plain dataclass) fall back to a repr() of the whole call, which renders them in full; pass a key_builder when they are large.

Timeouts

Coalescing ties your latency to a call you did not start. You join a flight that began before you arrived, you cannot see how long it has been running, and you cannot hurry it along — so if the upstream stalls, everyone who joined stalls with it. timeout bounds that wait:

@coalesce(timeout=2.0)
async def refresh_token() -> str:
    return await auth.post("/token")

Three things to know about it:

• The deadline belongs to the caller, not to the flight. Giving up cancels your own wait and nothing else; the work carries on for everybody still waiting. It is cancelled only when the caller who gives up was the last one left.
• It applies to whoever calls, first arrival or not. The caller who happens to start the flight is bound by the same deadline as the ones who join it.
• The clock starts when your call joins, not when the flight took off. A caller arriving four seconds in with timeout=2 waits until second six. timeout is not a cap on the work itself — put that inside the function, where your HTTP client or database driver can enforce it.

Giving up raises asyncio.TimeoutError. Catch it by that name: on Python 3.10 it is not the built-in TimeoutError, so except TimeoutError will not catch it there.

Coalescer.do() has no timeout argument, because its keyword arguments are forwarded to your function and timeout is a name real functions use. Set a deadline at the call site instead — it behaves exactly the same way:

await asyncio.wait_for(group.do("article:1", load_article, 1), 2.0)

What it guarantees

Cancelling one caller does not affect the others. The work runs in a task owned by the group, not in whichever coroutine happened to arrive first, so a request timeout or a disconnected client takes down only that caller's wait. This is the failure mode that naive implementations get wrong: if the first caller executes the function inline, its cancellation propagates to everyone waiting behind it. A timeout is the same story: it cancels the caller who set it, never the shared work while somebody else is still waiting on it.

Work stops when nobody is left waiting. If every caller is cancelled, the shared task is cancelled too — no orphaned queries running for results no one will read.

Failures are shared, not remembered. An exception reaches every waiter of that flight, then the key is retired; the next call runs the function again. coalescer never caches anything, including errors.

It is not a cache. A key exists only while its work is in flight. Two calls that do not overlap in time never share a result — put a cache in front for that.

Two details worth knowing before you deploy it:

• The result object is shared. All waiters get the same object, so mutating it affects everyone. Same as any memory cache.
• contextvars come from the first caller. The shared task inherits the context of whoever started the flight, so a request id or tracing span inside the coalesced call belongs to that caller. Unavoidable when one call serves many.

Coming from Go's singleflight

If you already know golang.org/x/sync/singleflight, here is the mapping:

Roadmap

• A threaded implementation for synchronous code at coalescer.sync, sharing the key logic.
• An optional Redis backend, so coalescing also holds across processes. The local group always deduplicates in-process first; the backend only decides which process runs the work.

Development

uv sync
uv run pytest
uv run ruff check . && uv run ruff format --check . && uv run mypy

The examples double as documentation, so CI runs them too:

uv run python examples/articles.py
uv run python examples/token_refresh.py

License

MIT