Metadata-Version: 2.4
Name: coalescer
Version: 0.1.0
Summary: Collapse concurrent duplicate async calls into a single execution — singleflight for asyncio: token refreshes, upstream calls, cache stampedes
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,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

[![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 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.

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

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

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

```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 flights are in the air.

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

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

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

**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:

| `golang.org/x/sync/singleflight` | `coalescer` |
|---|---|
| `Group.Do(key, fn)` | `await group.do(key, fn, *args, **kwargs)` |
| `Group.Forget(key)` | `group.forget(key)` |
| `Group.DoChan` | not needed — wrap `do()` in `asyncio.wait_for` or a `TaskGroup` |
| returns `shared bool` | not exposed |
| no cancellation story | cancellation is per-caller, with the shared task refcounted |

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

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

## License

MIT
