Metadata-Version: 2.4
Name: oneflight
Version: 0.1.0
Summary: Flexible sync and async singleflight: dedupe concurrent calls into one in-flight execution per key.
Keywords: singleflight,deduplication,cache,cache-stampede,async,asyncio,concurrency,threading,memoization
Author: Eugene Liukin
Author-email: Eugene Liukin <eugeneliukin.dev@gmail.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
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
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Project-URL: Homepage, https://github.com/eugeneliukin/oneflight
Project-URL: Repository, https://github.com/eugeneliukin/oneflight
Project-URL: Issues, https://github.com/eugeneliukin/oneflight/issues
Project-URL: Changelog, https://github.com/eugeneliukin/oneflight/blob/main/CHANGELOG.md
Description-Content-Type: text/markdown

# oneflight

**English** · [Русский](docs/README.ru.md) · [中文](docs/README.zh.md) · [Italiano](docs/README.it.md) · [Français](docs/README.fr.md)

[![PyPI version](https://img.shields.io/pypi/v/oneflight.svg)](https://pypi.org/project/oneflight/)
[![Python versions](https://img.shields.io/pypi/pyversions/oneflight.svg)](https://pypi.org/project/oneflight/)
[![CI](https://github.com/eugeneliukin/oneflight/actions/workflows/ci.yml/badge.svg)](https://github.com/eugeneliukin/oneflight/actions/workflows/ci.yml)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://github.com/eugeneliukin/oneflight/blob/main/LICENSE)
[![Typed](https://img.shields.io/badge/types-strict-blue.svg)](https://github.com/eugeneliukin/oneflight)

**Flexible sync and async singleflight for Python.**
Collapse concurrent calls that share the same key into a single in-flight execution, then hand the
result to every caller. A tiny, dependency-free, fully typed building block for **cache-stampede
protection** and request deduplication — in both threaded and `asyncio` code.

```python
# 100 concurrent callers, one actual execution
value, shared = await flights.do("user:42", load_user, 42)
```

## Table of contents

- [Why](#why)
- [Features](#features)
- [Installation](#installation)
- [Quickstart](#quickstart)
  - [Synchronous](#synchronous)
  - [Asynchronous](#asynchronous)
  - [Decorators](#decorators)
- [Concepts](#concepts)
  - [Dedup, not cache](#dedup-not-cache)
  - [The `shared` flag](#the-shared-flag)
  - [`forget`](#forget)
- [API reference](#api-reference)
- [FastAPI integration](#fastapi-integration)
- [How it works](#how-it-works)
- [Development](#development)
- [License](#license)

## Why

When many callers ask for the same expensive thing at the same time — a cache miss under load, a
burst of identical HTTP requests, a hot database row — you usually want the work to happen **once**
and be shared, not stampede your backend N times. That is exactly what singleflight does: the first
caller for a key runs the function; everyone else with the same key, arriving while it is still
running, waits and receives the same result.

`oneflight` gives you this for both worlds with one consistent surface:

| | Class | Runtime |
| --- | --- | --- |
| Threads | `SingleFlight` | `threading` |
| asyncio | `AsyncSingleFlight` | `asyncio` |

## Features

- **Sync and async**, each optimized for its own model (a lock for threads, cooperative scheduling
  for `asyncio` — no lock overhead where it is not needed).
- **Three ways to use it**: the explicit `do(key, fn, *args)` method, a per-group `wrap` decorator,
  and the standalone `@singleflight` / `@async_singleflight` decorators.
- **Shares results and exceptions** with every waiter; the function runs once.
- **Cancellation-safe** (async): cancelling one waiter never cancels the shared work or the others.
- **`forget(key)`** to evict an in-flight call and invalidate proactively.
- **Fully typed** (`py.typed`, `mypy --strict` clean) and **zero runtime dependencies**.

## Installation

```sh
pip install oneflight
# or
uv add oneflight
```

Requires Python 3.10+.

## Quickstart

### Synchronous

```python
from oneflight import SingleFlight

flights = SingleFlight()


def load_user(user_id: int) -> dict[str, str]: ...  # expensive DB / network call


# Concurrent threads calling this with the same key run load_user once and share the result.
user, shared = flights.do(f"user:{user_id}", load_user, user_id)
```

### Asynchronous

```python
import asyncio
from oneflight import AsyncSingleFlight

flights = AsyncSingleFlight()


async def load_user(user_id: int) -> dict[str, str]: ...  # expensive DB / network call


async def main() -> None:
    # 50 concurrent awaits, one load_user execution.
    results = await asyncio.gather(*(flights.do(f"user:{uid}", load_user, uid) for uid in [42] * 50))
    assert all(value == results[0][0] for value, _ in results)
```

### Decorators

Wrap a function so every call is automatically deduplicated. `key` maps the arguments to a hashable
key (defaults to the call's positional and keyword arguments).

```python
from oneflight import singleflight, async_singleflight


@singleflight(key=lambda user_id: user_id)
def load_user(user_id: int) -> dict[str, str]: ...


@async_singleflight(key=lambda user_id: user_id)
async def load_user_async(user_id: int) -> dict[str, str]: ...
```

For a shared key space across several functions, build one group and reuse its `wrap`:

```python
from oneflight import AsyncSingleFlight

flights = AsyncSingleFlight()


@flights.wrap(key=lambda user_id: f"user:{user_id}")
async def load_user(user_id: int) -> dict[str, str]: ...
```

## Concepts

### Dedup, not cache

`oneflight` deduplicates calls that **overlap in time**; it does not cache. As soon as the function
returns (or raises), the in-flight entry is removed, so the next call starts fresh. Pair it with an
actual cache when you want to remember results between waves:

```python
async def get_user(user_id: int) -> dict[str, str]:
    if (cached := cache.get(user_id)) is not None:
        return cached
    user, _ = await flights.do(f"user:{user_id}", load_and_cache_user, user_id)
    return user


async def load_and_cache_user(user_id: int) -> dict[str, str]:
    user = await load_user(user_id)
    cache.set(user_id, user)
    return user
```

Keep the write **inside** the deduplicated function. `load_and_cache_user` runs once per key during a
stampede, so `cache.set` fires exactly once; every other concurrent caller just reuses the result. If
you instead wrote to the cache after `do` returns, every waiter would repeat it — N redundant writes.

### The `shared` flag

`do` returns a `Flight[T]` — a `(value, shared)` tuple. `shared` tells you whether the value was
handed to more than one caller:

```python
value, shared = flights.do(key, fn)
```

- `shared is False` — the function ran just for you.
- `shared is True` — the result was reused by other callers (you were a waiter, or others joined
  while you were the owner).

Most callers ignore it (`value, _ = flights.do(...)`). It matters when the function yields a
**non-shareable** resource (a single-use token, an exclusive handle) that must not be given to two
callers — in that case, re-run when `shared` is `True`.

### `forget`

`forget(key)` evicts the current in-flight call so **future** callers start a new execution instead
of joining the running one. Use it when the in-flight call is known to be stale (data changed under
it) or stuck. Callers already attached to the old call still receive its result.

```python
flights.forget(f"user:{user_id}")  # next do() for this key runs fresh
```

## API reference

Both `SingleFlight` and `AsyncSingleFlight` share the same surface (the async methods are
coroutines):

| Member | Description |
| --- | --- |
| `do(key, fn, *args, **kwargs) -> Flight[T]` | Run `fn` once per in-flight `key`; returns `(value, shared)`. Waiters share the value; exceptions propagate to all. |
| `wrap(key=None) -> decorator` | Decorate a function so its calls are deduplicated. `key` computes the key from the arguments. |
| `forget(key) -> None` | Evict the in-flight entry for `key`. |

Module-level helpers, each backed by its own group:

| Member | Description |
| --- | --- |
| `@singleflight` / `@singleflight(key=...)` | Deduplicate a sync function. |
| `@async_singleflight` / `@async_singleflight(key=...)` | Deduplicate an async function. |
| `Flight[T]` | Type alias for `tuple[T, bool]` — the `(value, shared)` result. |

Keys must be **hashable**; a non-hashable key raises `TypeError`.

## FastAPI integration

A classic use case: protect an endpoint from a **cache stampede**. Under a burst of concurrent
requests for the same resource, only one upstream/DB call is made and every request shares it.

Own the group in the app's [`lifespan`](https://fastapi.tiangolo.com/advanced/events/) — it is
created on startup, stored on `app.state`, and injected into endpoints as a dependency (no module
globals):

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, Request
from oneflight import AsyncSingleFlight


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    app.state.flights = AsyncSingleFlight()
    yield


app = FastAPI(lifespan=lifespan)


def get_flights(request: Request) -> AsyncSingleFlight:
    return request.app.state.flights


Flights = Annotated[AsyncSingleFlight, Depends(get_flights)]


async def load_product(product_id: int) -> dict[str, object]: ...  # expensive DB / upstream call


@app.get("/products/{product_id}")
async def get_product(product_id: int, flights: Flights) -> dict[str, object]:
    product, _ = await flights.do(f"product:{product_id}", load_product, product_id)
    return product
```

When a product changes, evict it so the next reader does not join an in-flight stale load:

```python
@app.post("/products/{product_id}")
async def update_product(product_id: int, flights: Flights) -> None:
    ...  # write to the database
    flights.forget(f"product:{product_id}")
```

> One group per process deduplicates within that worker. Across multiple worker processes each has
> its own group; for cross-process coordination, put a shared cache (e.g. Redis) in front.

## How it works

- The group keeps a map of `key -> in-flight call`. The first caller for a key becomes the **owner**,
  creates the entry, and runs the function. Later callers for the same key find the entry and become
  **waiters**.
- Completion is signalled with an event (`threading.Event` in the sync group, `asyncio.Event` in the
  async one). Waiters block on it, then read the shared value or re-raise the shared exception.
- When the owner finishes, the entry is removed so the next wave starts fresh — an identity check
  guards against removing an entry that `forget` already replaced.
- The async group needs no lock (a single event loop serialises access); the sync group uses one
  lock for the map. Both share a small abstract base.

## Development

The task runner is [`just`](https://github.com/casey/just); git hooks run through
[`prek`](https://github.com/j178/prek) (a drop-in `pre-commit` runner).

```sh
just install       # sync the dev environment (uv)
just hooks         # install git hooks (pre-commit + pre-push)
just check         # fmt + lint + typecheck + test (the CI gate)
just test          # run the test suite
just build         # build the sdist and wheel
```

CI runs the exact same pre-commit hooks and the test suite across Python 3.10–3.14. Releases publish
to PyPI via [Trusted Publishing](https://docs.pypi.org/trusted-publishers/) when a GitHub Release is
published.

## License

[MIT](https://github.com/eugeneliukin/oneflight/blob/main/LICENSE) © Eugene Liukin
