mobius-idempotence-library

Idempotency-Key middleware for FastAPI, with pluggable storage backends.

Protects POST/PATCH-style endpoints (payments, order creation, anything non-idempotent by nature) against duplicate side effects when a client retries a request -- due to a timeout, a dropped connection, or an at-least-once delivery mechanism upstream.

Features

• Idempotency-Key enforcement following the shape of the common Idempotency-Key header convention used by Stripe and similar APIs.
• Request fingerprinting: a key reused with a different request body is rejected (422), not silently replayed.
• In-flight collision detection: a second concurrent request with the same key while the first is still processing gets 409, not a race.
• Pluggable backends: ships with RedisBackend (distributed, atomic via single-round-trip Lua scripts) and MemoryBackend (in-process, for tests or single-instance deployments). Implement StorageBackend for anything else (Postgres, DynamoDB, ...).
• Fail-open or fail-closed, your choice, when the backend is unreachable.
• Fully async, doesn't block the event loop.

Install

pip install mobius-idempotence-library          # in-memory backend only
pip install mobius-idempotence-library[redis]   # + Redis backend

Quick start

from fastapi import FastAPI
from pydantic import BaseModel

from mobius_idempotence import IdempotencyConfig, IdempotencyManager, idempotent
from mobius_idempotence.backends.redis import RedisBackend

app = FastAPI()

backend = RedisBackend("redis://localhost:6379/0")
manager = IdempotencyManager(backend, IdempotencyConfig(fail_open=False))


class PaymentRequest(BaseModel):
    amount: float
    currency: str


@app.post("/payments", status_code=201)
@idempotent(manager, status_code=201)
async def create_payment(payload: PaymentRequest):
    return await charge_the_card(payload)  # your business logic, nothing else


@app.on_event("shutdown")
async def shutdown():
    await manager.aclose()

That's the entire integration. @idempotent(manager) handles:

• validating the Idempotency-Key header
• fingerprinting the request body, so a key reused with a different body is rejected (422)
• detecting an in-flight duplicate request (409)
• replaying the cached response verbatim on retry, with no re-run of your handler
• releasing the lock if your handler raises, so the client can safely retry with the same key

No Depends(...), no manual cache lookup, no manual save/release calls in the handler.

Callers must send the same Idempotency-Key header (and the same request body) on retry:

curl -X POST http://localhost:8000/payments \
  -H "Idempotency-Key: 3f29c1a2-..." \
  -H "Content-Type: application/json" \
  -d '{"amount": 10.5, "currency": "USD"}'

What gets cached: the handler's return value (dict, list, Pydantic model, or dataclass), encoded the same way FastAPI itself would encode it. If you need to return a raw Response / StreamingResponse, or want custom caching logic, use the lower-level IdempotencyGuard dependency instead -- see examples/payments_app_manual.py.

Writing your own backend

Subclass mobius_idempotence.StorageBackend and implement five async methods: check_or_lock, save_response, release, ping, aclose. See src/mobius_idempotence/backends/memory.py for the simplest possible reference implementation.

Configuration

IdempotencyConfig is a plain dataclass -- wire it up to whatever config system your app already uses (env vars, Pydantic Settings, a YAML file):

Logging

The library never calls logging.basicConfig() or attaches its own handlers -- like any well-behaved library, it just emits log records and leaves handler/format/output configuration to your app. If you see nothing, that's almost always because your app hasn't configured logging at all, or has the root level above DEBUG/INFO.

Logger names:

• mobius_idempotence -- the core decision log (see below)
• mobius_idempotence.redis -- Redis connection-level errors only

What gets logged, at what level:

Turn it on without touching your app's global logging config:

import logging

logging.getLogger("mobius_idempotence").setLevel(logging.DEBUG)
logging.getLogger("mobius_idempotence").addHandler(logging.StreamHandler())

Or, if your app already calls logging.basicConfig(...) / dictConfig(...), just make sure the level for this logger (or the root logger) is DEBUG or INFO -- no extra wiring needed:

logging.basicConfig(level=logging.INFO)  # mobius_idempotence INFO/WARNING/ERROR now visible

Every log line includes the idempotency key so you can grep/filter a single request's lifecycle across a busy log stream.

Testing

pip install -e ".[dev]"
pytest

License

MIT