Metadata-Version: 2.5
Name: coalescer
Version: 0.3.0
Summary: Collapse concurrent duplicate calls into a single execution — singleflight for asyncio and threads
Project-URL: Homepage, https://github.com/roman-postnov/coalescer
Project-URL: Source, https://github.com/roman-postnov/coalescer
Project-URL: Issues, https://github.com/roman-postnov/coalescer/issues
Project-URL: Changelog, https://github.com/roman-postnov/coalescer/blob/master/CHANGELOG.md
Author-email: Roman Postnov <postnov.romen@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: asyncio,cache-stampede,concurrency,deduplication,dogpile,rate-limiting,request-coalescing,singleflight,threading,thundering-herd
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown

# coalescer

![One hundred concurrent callers converging into a single query](https://raw.githubusercontent.com/roman-postnov/coalescer/master/assets/social-preview-small.png)

[![PyPI](https://img.shields.io/pypi/v/coalescer.svg)](https://pypi.org/project/coalescer/)
[![Python](https://img.shields.io/pypi/pyversions/coalescer.svg)](https://pypi.org/project/coalescer/)
[![CI](https://github.com/roman-postnov/coalescer/actions/workflows/ci.yml/badge.svg)](https://github.com/roman-postnov/coalescer/actions/workflows/ci.yml)

**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`.

```python
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

```bash
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:

```python
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:

```python
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:

```python
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:

| Order | Cache hits | On a miss (100 concurrent callers) |
|---|---|---|
| cache outer, `@coalesce` inner | Answered straight from the cache, no coordination | 100 cache reads, **1 database query**, 100 cache writes |
| `@coalesce` outer, cache inner | Duplicate hits also collapse into one cache read | **1 cache read**, **1 database query**, 1 cache write |

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 `GET`s.

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:

```python
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:

```python
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:

```python
group = Coalescer()


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


group.forget(...)
```

## Async options

```python
@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:

```python
@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:

```python
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:

| `golang.org/x/sync/singleflight` | asyncio API | sync API |
|---|---|---|
| `Group.Do(key, fn)` | `await group.do(key, fn, *args, **kwargs)` | `group.do(key, fn, *args, **kwargs)` |
| `Group.Forget(key)` | `group.forget(key)` | `group.forget(key)` |
| `Group.DoChan` | use `asyncio.wait_for` or a `TaskGroup` | use your thread executor |
| returns `shared bool` | not exposed | not exposed |
| no cancellation story | cancellation is per-caller; work is refcounted | work runs in the first caller's thread |

## 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

```bash
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:

```bash
uv run python examples/articles.py
uv run python examples/token_refresh.py
uv run python examples/threaded_articles.py
```

## License

MIT
