idempotency-kit

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.

Features

• 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_idempotent for zero-boilerplate integration

Installation

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+

Quick start

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)

How it works

1. Client provides key

POST /api/orders
Idempotency-Key: abc-123-def

2. Server checks cache

cached = await repo.get("order.create", "abc-123-def")
if cached:
    return cached.result  # Return immediately ✅

3. Server executes (if not cached)

order = await create_order(dto)
await repo.save(record)  # Cache result for future requests
return order

4. Concurrent requests handled

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

Use cases

• 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

Documentation

📚 Full Documentation

• 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

Development

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 preview

See CONTRIBUTING.md for the full guide.

License

Apache 2.0