fastapi-idempotency

Idempotency-Key middleware for FastAPI and Starlette. Processes each write exactly once and replays the same response to retries, safely under concurrency, with a storage backend you control.

pip install fastapi-idempotency

from fastapi import FastAPI
from fastapi_idempotency import IdempotencyMiddleware

app = FastAPI()
app.add_middleware(IdempotencyMiddleware)

That's enough to try it. For anything beyond a single worker process, plug in a persistent Store — the in-memory default does not survive a restart or a second worker.

Why

A double-click, a client retry, or a network timeout followed by a resend all turn one intended write into two requests. This middleware recognizes the second one and returns the first one's response without running your endpoint again.

Two ways a request is recognized as a duplicate:

A write that runs longer than its lease (default 60s) keeps renewing it every heartbeat_seconds (default 20s) for as long as it's actually running, so a slow request is never mistaken for an abandoned one and duplicated.

What it will not do for you

This package guarantees that the middleware's own bookkeeping is consistent: exactly one request wins the claim, and the response it produces is what every duplicate gets back. It cannot make your endpoint's own side effects (a database write, a call to a payment provider) atomic with that bookkeeping unless they share the same transaction — if you need that guarantee for something like a payment, write the idempotency record in the same database transaction as the charge, using your own Store implementation, rather than relying on a generic HTTP layer.

Storage backends

from fastapi_idempotency import IdempotencyMiddleware, MemoryStore

app.add_middleware(IdempotencyMiddleware, store=MemoryStore())

# TORTOISE_ORM config
{"apps": {"models": {"models": ["myapp.models", "fastapi_idempotency.stores.tortoise"]}}}

from fastapi_idempotency.stores.tortoise import TortoiseStore

app.add_middleware(IdempotencyMiddleware, store=TortoiseStore())

Writing a backend for something else (SQLAlchemy, Redis, a plain table in whatever you already use) means implementing the Store protocol — five small async methods, documented in fastapi_idempotency/store.py. Contributions for new backends are welcome.

Identifying the caller: scope

By default, requests are grouped by client IP. That's a reasonable fallback for anonymous traffic, but for a logged-in API you almost always want to group by session or user instead — otherwise the automatic window won't recognize two requests from the same person behind a shared IP (or a NAT) as unrelated, and worse, a raw session-cookie value that rotates on every token refresh will make an Idempotency-Key stop being recognized the moment the token rotates in the background. Give it a stable identifier instead:

from starlette.requests import Request


def scope(request: Request) -> str:
    session_id = request.cookies.get("session_id")
    return f"session:{session_id}" if session_id else f"ip:{request.client.host}"


app.add_middleware(IdempotencyMiddleware, scope=scope)

scope may be sync or async, and returning an empty string is fine — it just means "one shared bucket."

Configuration

from datetime import timedelta
from fastapi_idempotency import IdempotencyMiddleware

app.add_middleware(
    IdempotencyMiddleware,
    header_name="Idempotency-Key",
    methods=("POST", "PUT", "PATCH", "DELETE"),
    path_prefix="/api",  # only these paths are checked; None checks all
    automatic_window=timedelta(seconds=10),
    key_window=timedelta(hours=24),
    lease_seconds=60,  # how long an in-flight claim is held
    heartbeat_seconds=20,  # how often a slow request renews its lease
    wait_seconds=15,  # how long a duplicate waits before RequestInProgressError
    max_request_body_bytes=8 * 1024 * 1024,
    max_response_body_bytes=8 * 1024 * 1024,
)

Errors

Every failure the middleware itself detects is one of these; by default they become a {"detail": "..."} JSON response with the status code shown. Override the shape with on_error — it may be sync or async def, so it can route through your app's own error handler (await handle_error(...), an audit log, a trace) instead of building the response itself:

from fastapi.responses import JSONResponse
from fastapi_idempotency import ConflictingRequestError, IdempotencyError


def on_error(request, error: IdempotencyError):
    if isinstance(error, ConflictingRequestError):
        return JSONResponse({"error": {"code": "IDEMPOTENCY_KEY_REUSED"}}, status_code=422)
    return JSONResponse({"error": {"code": "IDEMPOTENCY_ERROR"}}, status_code=500)


app.add_middleware(IdempotencyMiddleware, on_error=on_error)

Observability: on_event

Optional, for logging or wiring into your own request tracing:

def on_event(stage: str, status: str, message: str) -> None:
    logger.debug("[%s/%s] %s", stage, status, message)


app.add_middleware(IdempotencyMiddleware, on_event=on_event)

What it actually guarantees

• The claim is atomic. Two requests racing for the same key never both proceed — verified under concurrency in the test suite, for both stores.
• Streaming responses are captured without losing bytes, even when they cross the size limit mid-stream: the client still receives every byte, the response is just not stored.
• 5xx responses and responses that set cookies are never stored — a server error should be retryable, and a stored Set-Cookie should never be replayed to a different request.
• A completed record that outlives its window becomes reclaimable again, not a permanent, incorrect "still in progress."
• Hop-by-hop headers are stripped before a response is stored, so a replay never carries stale transport metadata.

Development

uv sync --all-extras
uv run pytest
uv run ruff check .
uv run basedpyright

License

MIT