A FastAPI Idempotency Dependency with Redis

FastAPI splits the request and response halves of a middleware across two different extension points, which makes idempotency slightly more awkward here than in Express or Fastify — and much easier to get subtly wrong. This runbook implements the obligations from idempotency middleware & framework integration in async Python.

Prerequisites. fastapi, redis.asyncio, and an authentication dependency that has already placed a credential on request.state. You should also have read Redis cache-based deduplication for the store semantics.

Step 1 — Cache the raw body

Starlette’s request stream is single-consumption. Reading it in a dependency drains it, and the route’s Pydantic model then receives an empty body.

The single-consumption stream, and the one-line fix The ASGI receive channel yields the request body once. A dependency that awaits request.body drains it, so the route's Pydantic model receives nothing. Assigning the bytes back to request underscore body restores the cache Starlette itself reads from, and the route parses normally. ASGI receive yields the body once raw = await request.body() request._body = raw route parses normally Pydantic reads the restored cache Without the restore, every wrapped route returns 422 on a perfectly valid payload. 
from fastapi import Request

async def read_and_cache_body(request: Request) -> bytes:
    raw = await request.body()
    request._body = raw     # restore Starlette's own cache so the route still parses
    return raw

Assigning to request._body looks like reaching into internals because it is, but it is the documented behaviour of Request.body(), which populates exactly that attribute on first read. The assignment is only needed when something else has already consumed the stream; keeping it makes the dependency safe to compose in any order.

Where each half of the FastAPI integration lives An outer box represents the http middleware, which wraps everything. Inside it, the dependency runs first, reading and caching the body, computing the fingerprint and registering the key atomically, then storing the scoped key on request.state. Next the endpoint executes. On the way out, the middleware reads request.state and writes the completion or release. Arrows show the inbound path through the dependency to the endpoint and the outbound path back through the middleware. @app.middleware("http") — sees the Response dependency read + cache body fingerprint register atomically → request.state.idem_key endpoint Pydantic parses the cached body normally side effect happens here Response status + body bytes middleware finalises: complete on 2xx/4xx, release on 5xx and 429 The dependency cannot do this — it never sees the Response. request.state carries the key across.

Step 2 — The dependency

import hashlib, json
from fastapi import Depends, HTTPException, Request

KEY_RE = re.compile(r"\A[A-Za-z0-9_-]{16,255}\Z")

def canonical(raw: bytes) -> bytes:
    try:
        doc = json.loads(raw or b"null")
    except json.JSONDecodeError:
        return raw                                   # non-JSON: hash the bytes as sent
    return json.dumps(doc, sort_keys=True, separators=(",", ":"),
                      ensure_ascii=False).encode("utf-8")

async def idempotency_guard(request: Request) -> None:
    raw_key = request.headers.get("idempotency-key")
    if raw_key is None:
        raise HTTPException(400, {"code": "idempotency_key_required"})
    if not KEY_RE.match(raw_key):
        raise HTTPException(400, {"code": "idempotency_key_invalid"})

    body = await read_and_cache_body(request)
    fingerprint = hashlib.sha256(canonical(body)).hexdigest()
    scoped = f"idem:v1:{request.state.credential_id}:{request.url.path}:{raw_key}"

    try:
        hit = await request.app.state.idem.register_or_get(scoped, fingerprint)
    except RedisError:
        raise HTTPException(503, {"code": "idempotency_store_unavailable"},
                            headers={"Retry-After": "2"})

    if hit is None:
        request.state.idem_key = scoped              # middleware finalises this
        return

    if hit["fingerprint"] != fingerprint:
        raise HTTPException(422, {"code": "idempotency_key_payload_mismatch"})
    if hit["state"] == "pending":
        raise HTTPException(409, {"code": "idempotency_key_in_use"},
                            headers={"Retry-After": "1"})
    raise ReplayResponse(status=int(hit["status"]), body=hit["body"])
class ReplayResponse(Exception):
    def __init__(self, status: int, body: str) -> None:
        self.status, self.body = status, body

@app.exception_handler(ReplayResponse)
async def replay_handler(_request: Request, exc: ReplayResponse) -> Response:
    return Response(content=exc.body, status_code=exc.status,
                    media_type="application/json",
                    headers={"Idempotent-Replay": "true"})

Raising an exception to return a stored response is unusual but correct here: a dependency can only short-circuit by raising, and a dedicated exception type keeps the replay path out of the endpoint’s signature.

Step 3 — The finalising middleware

@app.middleware("http")
async def finalise_idempotency(request: Request, call_next):
    response = await call_next(request)
    scoped = getattr(request.state, "idem_key", None)
    if scoped is None:
        return response

    # StreamingResponse bodies must be drained to be stored; bounded and buffered.
    chunks = [chunk async for chunk in response.body_iterator]
    body = b"".join(chunks)

    try:
        if response.status_code >= 500 or response.status_code == 429:
            await request.app.state.idem.release(scoped)
        else:
            await request.app.state.idem.complete(scoped, response.status_code, body)
    except RedisError:
        logger.error("idempotency finalise failed", extra={"key": scoped})
        metrics.inc("idempotency_finalise_error_total")

    response.headers["Idempotent-Replay"] = "false"
    return Response(content=body, status_code=response.status_code,
                    headers=dict(response.headers), media_type=response.media_type)

Draining body_iterator is the price of this design and it is why streaming routes must be excluded. Register the middleware only for the router that carries idempotent endpoints, or gate on request.state.idem_key as above so unwrapped routes stream normally.

Step 4 — Wire it up

import redis.asyncio as aioredis
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    pool = aioredis.ConnectionPool.from_url(
        settings.REDIS_URL, max_connections=50, socket_timeout=0.2)
    app.state.idem = RedisIdempotencyStore(aioredis.Redis(connection_pool=pool))
    yield
    await app.state.idem.close()

app = FastAPI(lifespan=lifespan)

@app.post("/v1/charges", status_code=201,
          dependencies=[Depends(authenticate), Depends(idempotency_guard)])
async def create_charge(payload: ChargeIn) -> ChargeOut:
    return await charges.create(payload)

socket_timeout=0.2 is deliberate. Without it, a Redis instance that is reachable but wedged turns every request into a multi-second hang, and the deduplication layer becomes the availability problem it was meant to prevent.

Step 5 — Test under real concurrency

import asyncio, httpx, pytest

@pytest.mark.asyncio
@pytest.mark.parametrize("trial", range(50))
async def test_concurrent_duplicates(trial, app, redis, db):
    key = f"01HTEST{trial:018d}"
    payload = {"amount": 4500, "currency": "gbp"}

    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        barrier = asyncio.Event()

        async def one():
            await barrier.wait()                       # release all 64 together
            return await client.post("/v1/charges", json=payload,
                                     headers={"Idempotency-Key": key})

        tasks = [asyncio.create_task(one()) for _ in range(64)]
        await asyncio.sleep(0.02)
        barrier.set()
        responses = await asyncio.gather(*tasks)

    assert sum(r.status_code == 201 for r in responses) == 1
    assert sum(r.status_code == 409 for r in responses) == 63
    assert await db.fetchval("SELECT count(*) FROM charges WHERE key=$1", key) == 1

Run this against a containerised Redis, never a fake. The guarantee is a property of the real store’s atomicity, and an in-process mock provides that atomicity for free — proving nothing.

docker run --rm -d -p 6379:6379 --name idem-test redis:7-alpine
REDIS_URL=redis://localhost:6379 pytest -q tests/test_idempotency.py
docker rm -f idem-test

Failure scenarios and debugging

Failure Scenario Remediation Steps Observability Hooks
Dependency reads the body without restoring request._body, so every route receives an empty payload and Pydantic raises 422 Restore the cache as in step 1 and add a smoke test that posts a valid payload through the dependency and asserts a 201. http_responses_total{status="422"} jumping to 100% on wrapped routes immediately after deploy
Finalisation placed in the dependency’s teardown, where the response is not available, so keys never leave pending Move it to the http middleware and pass the key through request.state. Assert in a test that a completed key has state=completed. idempotency_pending_age_seconds climbing steadily with zero idempotency_finalise_total
A new redis.asyncio client per request, adding a TCP handshake to every call Create one client in lifespan and share it via app.state. Grep for aioredis.Redis( outside the lifespan function in CI. idempotency_registration_duration_ms p50 above 5 ms on a local Redis is the tell
Middleware drains a StreamingResponse on an export route, so the client waits for the whole file before receiving anything Exclude streaming routes: register the middleware on a sub-router, or return early when request.state.idem_key is unset. response_first_byte_seconds on streaming routes; a value equal to total duration means it was buffered
HTTPException raised by the guard is not stored, so a client sending a bad payload retries against validation forever Let the finalising middleware store 4xx responses — it already does, because exceptions still produce a Response by the time the middleware sees it. Verify with a test asserting a stored 422. idempotency_finalise_total{status="422"} should be non-zero
One client for the process, not one per request Two arrangements. A single Redis client created in the lifespan handler and shared through app state reuses pooled connections, so a registration costs one round trip. A client constructed per request opens a fresh TCP connection each time, adding a handshake that typically costs more than the deduplication check itself. shared client (lifespan) pooled connections, reused registration ≈ 1 round trip socket_timeout keeps a wedged node visible client per request TCP handshake every call handshake > the check it protects and the pool never warms

SRE / observability checklist

  1. idempotency_registration_duration_ms — Histogram of the Redis round trip. Alert if p99 exceeds 15 ms; with socket_timeout=0.2 a wedged instance shows here first.
  2. idempotency_finalise_error_total — Counter of failed completion writes. Page above 0.05% of wrapped requests.
  3. idempotency_finalise_total{status} — Counter by response status; confirms that 4xx is stored and 5xx/429 released.
  4. idempotency_store_unavailable_total — Counter of the fail-closed 503 path. Any sustained value is a Redis availability incident, not a client problem.
  5. idempotency_pending_age_seconds — Gauge of the oldest pending key. Alert above 120 s.
  6. idempotency_key log field — Bound into the structured logger for the request so every line of a failed call carries it.