mobius-idempotence-library 0.1.1

Idempotency-Key middleware for FastAPI, with pluggable Redis / in-memory storage backends.

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):

Field	Default	Meaning
lock_ttl	60	Seconds an in-progress lock lives before being considered abandoned
response_ttl	86400	Seconds a finished response stays cached and replayable
key_prefix	"idemp:"	Storage key namespace
fail_open	True	If the backend is unreachable: process anyway (True) or 503 (False)
max_key_len	255	Reject longer Idempotency-Key headers

Testing

pip install -e ".[dev]"
pytest

License

MIT