Production-ready idempotency library for async Python applications.
Ensure operations execute exactly once, even when called multiple times with the same idempotency key. Built for production microservices with graceful degradation, collision handling, and observability.
Tip
Building this with an AI assistant? Hand it one page instead of the whole site: the complete API surface, the rules that break code when they are broken — what the key actually identifies, what two concurrent callers really do, how a TTL in seconds is rounded — the mistakes models make with this API, and a map of which page to fetch for the rest. Every docs page is also served as raw Markdown at its own URL, and a Copy page button at the top of each one hands it straight to a chat window.
- Clean Architecture — core domain separated from infrastructure
- Protocol-Based — easy to swap storage backends (Redis, custom)
- Type-Safe — full type hints with Pydantic validation
- Async First — built for asyncio applications
- Graceful Degradation — high availability over strict exactly-once
- In-flight Reservation — a retry that arrives while the original is still running waits for its result or gets a 409; the action runs once
- Request Fingerprints — name the parameters that identify a request, and a key reused for a different one is refused instead of replayed
- Observability — built-in metrics (hits, misses, collisions, latency)
- Bulk Operations — efficient
get_many,save_many,delete_many - Redis Cluster Compatible — non-transactional pipelines
- Decorator Pattern —
@async_idempotentfor zero-boilerplate integration
pip install idempotency-kit
# With Redis support (recommended)
pip install idempotency-kit[redis]
# With Dishka DI
pip install idempotency-kit[dishka,redis]Requirements: Python 3.11+, Redis 6+
from idempotency_kit import AsyncIdempotencyCoordinator, PydanticResultAdapter, async_idempotent
class CreateOrderUseCase:
def __init__(self, uow: AsyncUnitOfWork, coordinator: AsyncIdempotencyCoordinator):
self._uow = uow
self.coordinator = coordinator
@async_idempotent(
operation="order.create",
adapter=PydanticResultAdapter(OrderDTO),
)
async def execute(
self,
dto: CreateOrderDTO,
idempotency_key: str | None = None,
) -> OrderDTO:
"""Create order - idempotency handled automatically."""
async with self._uow.transaction() as tx:
# Your business logic - no idempotency code needed!
order = await tx.orders.create(dto.items, dto.total)
await tx.outbox.create(OrderCreatedEvent(order_id=order.id))
return OrderDTO.from_entity(order)Pass idempotency_key to downstream services for distributed idempotency:
# Orchestrate multiple services with same key
await identity_service.create_user(..., idempotency_key=idempotency_key)
await payment_service.charge(..., idempotency_key=idempotency_key)POST /api/orders
Idempotency-Key: abc-123-def
cached = await repo.get("order.create", "abc-123-def")
if cached:
return cached.result # Return immediately ✅order = await create_order(dto)
await repo.save(record) # Cache result for future requests
return orderIf two requests arrive while the first is still executing:
- First request: reserves the key → execute → write the result over the reservation ✅
- Second request: finds the reservation → waits for the first result → return ✅ (or a 409 with
in_flight="raise")
Both requests get the same result, and the business logic ran once.
- HTTP APIs — ensure POST/PUT requests are idempotent
- Background Jobs — prevent duplicate processing on retries
- Event Consumers — handle duplicate events gracefully
- Message Queues — at-most-once message processing
- For AI agents — the whole library on one page
- Quick Start — get started in 5 minutes
- User Guide — detailed usage and patterns
- Architecture — design principles
- API Reference — complete API docs
make install # uv sync --group dev
make check # ruff + mypy
make test-unit # unit tests (no Docker)
make test-integration # integration tests (Docker required)
make test # all tests with coverage
make docs-serve # local docs previewSee CONTRIBUTING.md for the full guide.