coalescer

Collapse concurrent duplicate calls into a single execution. Hang one decorator on a function you already have: while a call is in flight, everyone asking for the same thing joins it instead of starting their own. The default API is for asyncio; synchronous threaded code lives at coalescer.sync.

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.

Synchronous threads

The same API is available for ordinary functions called from multiple threads:

from coalescer.sync import coalesce


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

Or drive the synchronous group directly:

from coalescer.sync import Coalescer

group = Coalescer()
article = group.do(f"article:{article_id}", load_article_from_db, article_id)

coalescer.sync coordinates threads you already use; it does not create any. The first caller for a key runs the function in its own thread, while the others block until it returns. Its decorator has the same key, key_builder, noself and group options as the async one.

There is deliberately no synchronous timeout option. Python cannot safely stop an arbitrary blocking function, and a deadline that applied only to followers would treat the first caller differently. Put the timeout on the database, HTTP, or other blocking operation itself.

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 function 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: both APIs call key_builder(func, *args, **kwargs), the convention cache decorators use, so one callable can be handed to a cache, async coalescer, or sync coalescer 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 keys can be joined right now.

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(...)

Async 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 calls share a flight when their arguments are of the same type and equal — lists, dicts, sets and tuples included, however deeply nested. A method is keyed per instance, so noself=True is what you want when callers build their own instance instead of sharing one.

An argument that is none of those and cannot be hashed — a plain dataclass, say — raises TypeError: pass a key_builder, a fixed key, or noself=True when it is the instance a decorated method was called on.

Async 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.

Those cancellation guarantees belong to the async API. Synchronous work belongs to the first caller's thread and cannot be cancelled by the group; followers only wait for what that thread produces.

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.
• In async code, 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

• 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
uv run python examples/threaded_articles.py

License

MIT