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.

Easy Python 3.9+ Aug 9, 2026 Reliability & rate limiting 15 views 0 copies

Python code

39 lines
Python 3.9+
import 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

stdout
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

  1. Replace the in-memory dict with Redis using `SET key value EX ttl NX` for distributed idempotency.
  2. 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

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Reliability & rate limiting

Related tutorials and quizzes for this topic.