Single-flight token refresh for
httpxandrequests: when hundreds or thousands of parallel requests all get a 401, yourrefresh()function is called exactly once.
import httpx
from singleflight_auth import SingleFlightAuth
def get_access_token() -> str:
return token_store.get("access")
def refresh_access_token() -> str:
resp = httpx.post(
"https://api.example.com/auth/refresh",
json={"refresh_token": token_store.get("refresh")},
)
resp.raise_for_status()
data = resp.json()
token_store.set("access", data["access_token"])
return data["access_token"]
auth = SingleFlightAuth(get_token=get_access_token, refresh=refresh_access_token)
client = httpx.Client(auth=auth, base_url="https://api.example.com")
# Even if unlimited parallel requests hit a 401, refresh is called exactly once.from singleflight_auth import AsyncSingleFlightAuth
async def async_refresh() -> str:
async with httpx.AsyncClient() as c:
resp = await c.post("https://api.example.com/auth/refresh", json={...})
resp.raise_for_status()
data = resp.json()
token_store.set("access", data["access_token"])
return data["access_token"]
auth = AsyncSingleFlightAuth(get_token=get_access_token, refresh=async_refresh)
async with httpx.AsyncClient(auth=auth) as client:
# Whether it's 10, 50, or 10,000 requests, they are all coordinated.
responses = await asyncio.gather(*[client.get("/api/data") for _ in range(500)])
# refresh() was called exactly once — all 500 got the fresh token.import requests
from singleflight_auth import RequestsSingleFlightAuth
auth = RequestsSingleFlightAuth(get_token=get_access_token, refresh=refresh_access_token)
session = requests.Session()
session.auth = auth
response = session.get("https://api.example.com/protected")| Library | Scope | Why it's different |
|---|---|---|
httpx (official) |
Auth.auth_flow pattern documented |
No lock/queue — DIY, no concurrency safety |
httpx-auth |
Full OAuth2 client (auth code, PKCE, client credentials, browser integration) | Heavy, spec-bound; 541K+ weekly downloads but doesn't fit custom refresh endpoints |
requests_oauth2client |
Full OAuth2/OIDC for requests |
Full spec implementation — overkill if you only want "call my refresh on 401" |
singleflight |
General call-coalescing (Go's groupcache port) | Not HTTP/auth-specific; no 401 detection or retry logic |
Our position: If you need full OAuth2 flows, use httpx-auth or requests_oauth2client — they're excellent. If you already have your own refresh logic (JWT endpoint, custom auth API, or even an OAuth2 token endpoint you call manually) and just want to prevent parallel 401s from stomping each other — this library is for you.
"Bring your own refresh logic — we handle the concurrency."
# For httpx users (sync + async):
pip install singleflight-auth[httpx]
# For requests users:
pip install singleflight-auth[requests]
# Both:
pip install singleflight-auth[httpx,requests]The core uses double-checked locking — a well-known concurrency pattern adapted for token refresh:
┌──────────────────────────────────────────┐
│ 50 requests hit 401 │
└──────────────────┬───────────────────────┘
│
┌──────────────────▼───────────────────────┐
│ Each remembers stale token T_stale │
└──────────────────┬───────────────────────┘
│
┌──────────────────▼───────────────────────┐
│ Acquire lock (one wins) │
│ threads: block │ coroutines: yield │
└──────────────────┬───────────────────────┘
│
┌──────────────────▼───────────────────────┐
│ Re-read current token T_current │
└───────┬──────────────────────┬───────────┘
│ │
T_current ≠ T_stale T_current == T_stale
│ │
┌───────▼──────┐ ┌────────▼──────────┐
│ Skip refresh │ │ YOU are the winner │
│ Use T_current│ │ Call refresh() │
└───────┬──────┘ └────────┬──────────┘
│ │
┌───────▼──────────────────────▼───────────┐
│ Release lock │
│ Retry request with fresh token │
└──────────────────────────────────────────┘
Result: N concurrent 401s → exactly 1 refresh() call
The other N-1 get the fresh token for free.
Key design decisions:
SyncCoordinatorusesthreading.Lock— threads block while waiting.AsyncCoordinatorusesasyncio.Lock— coroutines yield without blocking the event loop.- The two are never mixed:
httpx.Clientuses the sync path,httpx.AsyncClientuses the async path.
from singleflight_auth import SingleFlightAuth
auth = SingleFlightAuth(
get_token: Callable[[], str], # Returns the current token
refresh: Callable[[], str], # Fetches a new token, saves it, returns it
is_unauthorized: Callable[ # Optional: customize 401 detection
[httpx.Response], bool
] = lambda r: r.status_code == 401,
max_retries: int = 1, # Max retry attempts after refresh
)from singleflight_auth import AsyncSingleFlightAuth
auth = AsyncSingleFlightAuth(
get_token: Callable[[], str],
refresh: Callable[[], Awaitable[str]], # Must be an async function
is_unauthorized: Callable[[httpx.Response], bool] = ...,
max_retries: int = 1,
)from singleflight_auth import RequestsSingleFlightAuth
auth = RequestsSingleFlightAuth(
get_token: Callable[[], str],
refresh: Callable[[], str],
is_unauthorized: Callable[[requests.Response], bool] = ...,
max_retries: int = 1,
)
session = requests.Session()
session.auth = auth| Exception | When |
|---|---|
RefreshFailedError |
Your refresh() callable raised an exception, or returned an empty/None token |
MaxRetriesExceededError |
Still getting 401 after max_retries refresh attempts |
ReentrantRefreshError |
refresh() triggered the auth flow on the same client/session (would deadlock) |
NonReplayableBodyError |
Request body is a stream/generator/file that cannot be replayed for retry |
from singleflight_auth import (
RefreshFailedError,
MaxRetriesExceededError,
ReentrantRefreshError,
NonReplayableBodyError,
)
try:
response = client.get("/protected")
except RefreshFailedError as e:
# refresh() raised — log out the user, redirect to login
logger.error(f"Token refresh failed: {e.__cause__}")
except MaxRetriesExceededError:
# Server keeps returning 401 even after refresh — something is very wrong
logger.error("Max retries exceeded, giving up")
except ReentrantRefreshError:
# refresh() accidentally used the same client — fix your refresh() code
logger.error("refresh() must use a separate client!")
except NonReplayableBodyError:
# Streaming upload got a 401 — buffer the body or handle differently
logger.error("Cannot retry streaming uploads")The refresh() callable must never use the same httpx.Client / httpx.AsyncClient / requests.Session that has this auth handler attached. If it does, and the refresh endpoint also returns 401, the coordinator will detect the reentrant lock acquisition and raise ReentrantRefreshError.
# Correct — uses a standalone httpx.post() without auth
def refresh() -> str:
resp = httpx.post("https://auth.example.com/token", json={...})
return resp.json()["access_token"]
# Wrong — reuses the client that has SingleFlightAuth attached
def refresh() -> str:
resp = client.post("https://auth.example.com/token") # DANGER: same client!
return resp.json()["access_token"]The single-flight guarantee only works when all requests share the same auth instance. Each instance has its own lock, so creating a new SingleFlightAuth(...) per request defeats the entire purpose.
# Correct — one instance shared everywhere
auth = SingleFlightAuth(get_token=..., refresh=...)
client = httpx.Client(auth=auth)
# Wrong — new instance per request means N refreshes instead of 1
for url in urls:
auth = SingleFlightAuth(get_token=..., refresh=...) # WRONG
httpx.get(url, auth=auth)If the request body is a generator, file handle, or other one-shot stream, it is consumed on the first send and cannot be replayed. Instead of silently sending an empty body, NonReplayableBodyError is raised. Buffer the body into bytes before sending:
# Correct — body is bytes, safe to retry
data = my_file.read()
client.post("/upload", content=data)
# Risky — generator body cannot be replayed on 401
client.post("/upload", content=my_generator())Intentionally out of scope for v0.1:
| What | Why |
|---|---|
| No OAuth2 flow implementation | Bring your own refresh() logic — we don't dictate your auth scheme |
| No token storage | Manage tokens yourself via get_token/refresh callbacks |
| Single-process only | Uses in-process locks (threading.Lock / asyncio.Lock); does not work across multiple processes or machines (e.g., Gunicorn workers) |
No aiohttp support |
httpx + requests only for v0.1 |
| Reactive only | Refreshes on 401; no proactive TTL-based refresh |
| Stream bodies not retried | Generator/file/stream request bodies raise NonReplayableBodyError on 401 instead of silently losing data |
See CONTRIBUTING.md for development setup and guidelines.
# Quick start for contributors:
git clone https://github.com/alibeg-begow/singleflight_auth.git
cd singleflight_auth
uv sync --all-extras --dev
uv run pytest -v
uv run mypy src --strict
uv run ruff check .This project is licensed under the MIT License.