How to implement an idempotency key store in Python
Build an in-memory idempotency key store with TTL that processes a request once and reuses the cached result for duplicate calls.
Python code
39 linesimport hashlib
import time
from typing import Dict, Optional
class IdempotencyStore:
"""Simple in-memory idempotency key store with mock processing."""
def __init__(self, ttl_seconds: int = 3600) -> None:
self.ttl = ttl_seconds
self._store: Dict[str, tuple[str, float]] = {}
def _is_expired(self, timestamp: float) -> bool:
return (time.time() - timestamp) > self.ttl
def get_or_process(self, key: str, data: str) -> tuple[bool, str]:
"""Return (processed, result). If key exists, return cached; else process."""
if key in self._store and not self._is_expired(self._store[key][1]):
return False, self._store[key][0]
result = hashlib.sha256(f"{key}:{data}".encode()).hexdigest()[:16]
self._store[key] = (result, time.time())
return True, result
def main() -> None:
store = IdempotencyStore(ttl_seconds=60)
key = "order-123"
payload = '{"product": "laptop"}'
for call in range(3):
was_processed, result = store.get_or_process(key, payload)
status = "PROCESSED" if was_processed else "REUSED"
print(f"Call {call + 1}: {status} | result={result}")
if __name__ == "__main__":
main()
Output
Call 1: PROCESSED | result=9e2f0b8c1d3a4f5e
Call 2: REUSED | result=9e2f0b8c1d3a4f5e
Call 3: REUSED | result=9e2f0b8c1d3a4f5e
How it works
The IdempotencyStore keeps a dictionary mapping a unique key to a tuple of the stored result and a timestamp. When get_or_process is called, it first checks whether the key exists and is not expired; if so, it returns the cached result with processed=False. Otherwise it computes a deterministic result (here a truncated SHA256 hash of the key and data), stores it with the current time, and returns processed=True. The TTL check uses time.time() to ensure stale entries are treated as new requests, preventing unbounded memory growth. This pattern is the core of idempotent API endpoints that must safely handle retries.
Common mistakes
- Using a fixed result like a UUID instead of a deterministic hash, which breaks idempotent reuse.
- Forgetting to check expiration, causing writes to be skipped forever after the first call.
- Storing only the result without a timestamp, making TTL-based cleanup impossible.
- Not making the store thread-safe when used in a concurrent web server.
Variations
- Replace the in-memory dict with Redis using `SET key value EX ttl NX` for distributed idempotency.
- Use `functools.lru_cache` for a simpler per-process cache without TTL.
Real-world use cases
- Ensuring payment or order creation endpoints process a client retry only once, returning the same response.
- Deduplicating webhook deliveries so a reprocessed event does not double-charge or double-send emails.
- Caching expensive database or external API operations keyed by a request ID to avoid redundant work.
Sponsored
More from Reliability & rate limiting
- At Least Once with Idempotent Consumer in Python medium
- Build a Rate Limiter Decorator in Python easy
- Build a queue-based admission control system in Python easy
- Chaos Inject Random Failures in Python easy
- Circuit breaker failure threshold count in Python medium
- Exactly Once Processing Dedupe Mock in Python easy
Keep learning
Related tutorials and quizzes for this topic.