"""TTL-aware LRU cache. A small, dependency-free, thread-safe-ish (single-threaded design) cache combining least-recently-used eviction with per-entry time-to-live expiry. Design notes: - Ordering is tracked with an OrderedDict (move_to_end on access). - Expiry is checked lazily on get/peek; expired entries are dropped on access. - Capacity counts only live (non-expired) entries at insertion time; expired entries are purged opportunistically when capacity is exceeded. - A monotonic clock (time.monotonic) is used so wall-clock jumps don't break TTL. """ from __future__ import annotations import time from collections import OrderedDict from dataclasses import dataclass from typing import Any, Callable, Generic, Optional, TypeVar K = TypeVar("K") V = TypeVar("V") class CacheMiss(KeyError): """Raised by get_strict when a key is absent or expired.""" @dataclass class _Entry(Generic[V]): value: V expires_at: float # monotonic seconds; inf means no expiry class TTLCache(Generic[K, V]): def __init__( self, capacity: int, default_ttl: Optional[float] = None, clock: Callable[[], float] = time.monotonic, ) -> None: if capacity <= 0: raise ValueError("capacity must be positive") if default_ttl is not None and default_ttl <= 0: raise ValueError("default_ttl must be positive or None") self._capacity = capacity self._default_ttl = default_ttl self._clock = clock self._data: "OrderedDict[K, _Entry[V]]" = OrderedDict() # -- internals --------------------------------------------------------- def _now(self) -> float: return self._clock() def _purge_expired(self) -> None: # Drop expired entries from the front/back opportunistically. now = self._now() dead = [k for k, e in self._data.items() if e.expires_at <= now] for k in dead: del self._data[k] def _ttl_to_deadline(self, ttl: Optional[float]) -> float: if ttl is None: ttl = self._default_ttl if ttl is None: return float("inf") if ttl <= 0: raise ValueError("ttl must be positive") return self._now() + ttl # -- public API -------------------------------------------------------- def __len__(self) -> int: self._purge_expired() return len(self._data) def __contains__(self, key: K) -> bool: entry = self._data.get(key) if entry is None: return False if entry.expires_at <= self._now(): del self._data[key] return False return True def get(self, key: K, default: Any = None) -> Any: entry = self._data.get(key) if entry is None: return default if entry.expires_at <= self._now(): del self._data[key] return default self._data.move_to_end(key) return entry.value def get_strict(self, key: K) -> V: v = self.get(key, _MISSING) if v is _MISSING: raise CacheMiss(key) return v # type: ignore[return-value] def set(self, key: K, value: V, ttl: Optional[float] = None) -> None: deadline = self._ttl_to_deadline(ttl) if key in self._data: del self._data[key] self._data[key] = _Entry(value=value, expires_at=deadline) self._data.move_to_end(key) self._evict_if_needed() def _evict_if_needed(self) -> None: self._purge_expired() while len(self._data) > self._capacity: self._data.popitem(last=False) def delete(self, key: K) -> bool: if key in self._data: del self._data[key] return True return False def clear(self) -> None: self._data.clear() def keys(self): self._purge_expired() return list(self._data.keys()) def items(self): self._purge_expired() return [(k, e.value) for k, e in self._data.items()] def peek(self, key: K, default: Any = None) -> Any: """Return value without updating recency or triggering purge of others.""" entry = self._data.get(key) if entry is None or entry.expires_at <= self._now(): return default return entry.value _MISSING = object()