oneflight

English · Русский · 中文 · Italiano · Français

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.

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

Table of contents

• Why
• Features
• Installation
• Quickstart
  • Synchronous
  • Asynchronous
  • Decorators
• Concepts
  • Dedup, not cache
  • The shared flag
  • forget
• API reference
• FastAPI integration
• How it works
• Development
• 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

pip install oneflight
# or
uv add oneflight

Requires Python 3.10+.

Quickstart

Synchronous

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

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

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:

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:

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:

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.

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 — it is created on startup, stored on app.state, and injected into endpoints as a dependency (no module globals):

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:

@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; git hooks run through prek (a drop-in pre-commit runner).

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 when a GitHub Release is published.

License

MIT © Eugene Liukin