Metadata-Version: 2.4
Name: pyidempotent
Version: 0.1.0
Summary: Production-ready idempotency keys for FastAPI, Flask, Django and any Python function. Redis & memory backends, async-first.
Project-URL: Homepage, https://github.com/you/pyidempotent
Project-URL: Repository, https://github.com/you/pyidempotent
Author-email: pyidempotent <hello@example.com>
License: MIT
License-File: LICENSE
Keywords: exactly-once,fastapi,idempotency,payments,redis
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Requires-Python: >=3.9
Provides-Extra: dev
Requires-Dist: fastapi; extra == 'dev'
Requires-Dist: httpx; extra == 'dev'
Requires-Dist: pytest; extra == 'dev'
Requires-Dist: pytest-asyncio; extra == 'dev'
Requires-Dist: redis; extra == 'dev'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.100; extra == 'fastapi'
Requires-Dist: starlette>=0.27; extra == 'fastapi'
Provides-Extra: redis
Requires-Dist: redis>=5.0.0; extra == 'redis'
Description-Content-Type: text/markdown

# 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.

```python
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

```bash
pip install pyidempotent[redis,fastapi]
```

## Backends

```python
# 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

```python
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