import asyncio
from typing import Awaitable, Callable, Dict, TypeVar
T = TypeVar("T")
class SingleFlight:
def __init__(self) -> None:
self._inflight: Dict[str, asyncio.Future] = {}
async def do(self, key: str, factory: Callable[[], Awaitable[T]]) -> T:
existing = self._inflight.get(key)
if existing is not None:
return await existing
loop = asyncio.get_event_loop()
future: asyncio.Future = loop.create_future()
self._inflight[key] = future
try:
result = await factory()
except BaseException as exc:
if not future.done():
future.set_exception(exc)
raise
else:
if not future.done():
future.set_result(result)
return result
finally:
self._inflight.pop(key, None)
def inflight_count(self) -> int:
return len(self._inflight)
import asyncio
from typing import Dict
import httpx
from singleflight import SingleFlight
class PriceService:
def __init__(self, base_url: str) -> None:
self._base_url = base_url
self._client = httpx.AsyncClient(timeout=5.0)
self._flight = SingleFlight()
self.upstream_calls = 0
async def _fetch_upstream(self, symbol: str) -> float:
self.upstream_calls += 1
resp = await self._client.get(f"{self._base_url}/quote/{symbol}")
resp.raise_for_status()
return float(resp.json()["price"])
async def get_price(self, symbol: str) -> float:
symbol = symbol.upper()
async def factory() -> float:
return await self._fetch_upstream(symbol)
return await self._flight.do(symbol, factory)
async def warm(self, symbol: str, concurrency: int = 50) -> Dict[str, float]:
tasks = [self.get_price(symbol) for _ in range(concurrency)]
results = await asyncio.gather(*tasks)
return {"price": results[0], "upstream_calls": self.upstream_calls}
async def close(self) -> None:
await self._client.aclose()
from fastapi import APIRouter, Depends, HTTPException
import httpx
from price_service import PriceService
router = APIRouter()
_service = PriceService(base_url="https://quotes.internal")
def get_service() -> PriceService:
return _service
@router.get("/price/{symbol}")
async def read_price(symbol: str, service: PriceService = Depends(get_service)):
try:
price = await service.get_price(symbol)
except httpx.HTTPStatusError as exc:
raise HTTPException(status_code=502, detail=f"upstream error: {exc.response.status_code}")
except httpx.HTTPError:
raise HTTPException(status_code=504, detail="upstream unavailable")
return {"symbol": symbol.upper(), "price": price}
@router.on_event("shutdown")
async def shutdown() -> None:
await _service.close()
This snippet implements the single-flight pattern in Python's asyncio: when many callers ask for the same key at the same time, only one underlying computation runs and everyone else waits on and shares its result. It solves the thundering-herd problem, where a cache miss on a hot key would otherwise fire N identical database or upstream calls simultaneously.
In singleflight.py, the core idea is that the first caller for a key creates a shared asyncio.Future and stores it in self._inflight under that key, then runs the coroutine factory. Later callers that arrive while the computation is still running find the existing future and simply await it — they never invoke the factory. Because asyncio runs on a single thread and there are no await points between the dictionary lookup and the insertion, the check-and-insert in do is atomic without an explicit lock; a Lock is only needed if the code path yields between those steps. The finally block removes the key from _inflight so the next request after completion starts fresh, and both success and failure are propagated to every waiter via set_result and set_exception. Sharing the exception matters: all coalesced callers see the same failure rather than one succeeding while the rest hang.
The key trade-off is that a shared future means a shared fate — one slow or failing computation blocks all waiters, so this pattern coalesces identical work but does not add retries or per-caller timeouts. It also deduplicates only concurrent requests, not sequential ones; pairing it with a TTL cache handles that.
In price_service.py, SingleFlight guards an expensive upstream fetch. get_price wraps the real call in a factory closure passed to flight.do, so a burst of requests for the same symbol collapses into one HTTP round trip. warm demonstrates that concurrent get_price calls for one symbol trigger exactly one _fetch_upstream.
In routes.py, a FastAPI endpoint calls the shared service instance, meaning a spike of clients hitting /price/{symbol} naturally coalesces at the service layer. This is where the pattern shines: read-heavy endpoints with hot keys, cache stampede protection, and any idempotent-but-expensive lookup where duplicate concurrent work is pure waste.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
module Api
module V1
class UsersController < BaseController
def show
user = User.includes(:profile).find(params[:id])ETags for conditional requests and caching
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
