pyidempotent 0.1.0

Production-ready idempotency keys for FastAPI, Flask, Django and any Python function. Redis & memory backends, async-first.

pyidempotent

Production-ready idempotency keys for Python backends — in 5 lines.

Stop double charges, double orders, and duplicate webhooks. Exactly-once execution for FastAPI, Flask, Django, Celery — with Redis or in-memory store.

from fastapi import FastAPI, Request
from pyidempotent import idempotent
from pyidempotent.backends.redis import RedisBackend
import redis.asyncio as redis

app = FastAPI()
backend = RedisBackend(redis.from_url("redis://localhost"))

@app.post("/pay")
@idempotent(backend=backend, ttl=24*3600)
async def pay(request: Request, amount: int):
    # will run ONCE per Idempotency-Key
    return {"status": "charged", "amount": amount}

Send Idempotency-Key: 123e4567-e89b-12d3-a456-426614174000 header. Repeat the request — you get the cached response, no second charge.

---

Why pyidempotent?

• Async-first, works with sync too
• Zero framework lock-in — pure decorator
• Race-safe — Redis SET NX + atomic claim
• Request fingerprinting — 422 if same key but different body
• Processing state — returns 409 while first request still runs
• Pluggable backends — Redis (prod), Memory (tests)
• <400 LOC, no magic

Install

pip install pyidempotent[redis,fastapi]

Backends

# Production
from pyidempotent.backends.redis import RedisBackend
backend = RedisBackend(redis.from_url("redis://localhost"), prefix="idem:")

# Tests
from pyidempotent.backends.memory import MemoryBackend
backend = MemoryBackend()

FastAPI full example

from fastapi import FastAPI, Request, HTTPException
from pyidempotent import idempotent, IdempotencyConflict, IdempotencyProcessing
from pyidempotent.backends.redis import RedisBackend
import redis.asyncio as redis

app = FastAPI()
backend = RedisBackend(redis.from_url("redis://localhost"))

@app.exception_handler(IdempotencyConflict)
async def conflict_handler(_, exc):
    raise HTTPException(422, "Idempotency-Key already used with different payload")

@app.post("/orders")
@idempotent(backend=backend, key_header="Idempotency-Key")
async def create_order(request: Request, item: str):
    # your DB write here
    return {"order_id": "ord_123", "item": item}

How it works

1. Extract key from header
2. SET key {status:processing} NX EX ttl — atomic claim
3. Run your function
4. SET key {status:completed, response:...} EX ttl
5. Duplicates return cached response

Fingerprint = SHA256 of function arguments (excluding Request). Prevents accidental reuse.

License

MIT