Idempotency Key Store For Duplicate Requests

import hashlib
import time
from dataclasses import dataclass
from typing import Any, Dict, Optional


class IdempotencyConflictError(ValueError):
    """Raised when a key is reused with a different payload."""


@dataclass(frozen=True)
class Entry:
    payload_hash: str
    response: Any
    stored_at: float


def payload_fingerprint(payload: str) -> str:
    """Return a stable fingerprint for a request payload."""
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


class IdempotencyStore:
    """Replay recorded responses for duplicate keys; reject payload conflicts."""

    def __init__(self, ttl_s: float = 3600.0, clock: Any = time.monotonic) -> None:
        if ttl_s <= 0:
            raise ValueError("ttl_s must be positive")
        self._ttl_s = ttl_s
        self._clock = clock
        self._entries: Dict[str, Entry] = {}

    def execute(self, key: str, payload: str, produce: Any) -> Any:
        """Return the recorded response, or run produce and record it."""
        if not key:
            raise ValueError("idempotency key must be non-empty")
        fingerprint = payload_fingerprint(payload)
        existing = self._get_live(key)
        if existing is not None:
            if existing.payload_hash != fingerprint:
                raise IdempotencyConflictError(
                    f"key {key!r} was already used with a different payload"
                )
            return existing.response
        response = produce()
        self._entries[key] = Entry(
            payload_hash=fingerprint,
            response=response,
            stored_at=self._clock(),
        )
        return response

    def _get_live(self, key: str) -> Optional[Entry]:
        entry = self._entries.get(key)
        if entry is None:
            return None
        if self._clock() - entry.stored_at > self._ttl_s:
            del self._entries[key]
            return None
        return entry