from __future__ import annotations import time from typing import Any, Dict, Tuple, Optional class TTLCache: """Simple in-memory TTL cache (per-process). Not thread-safe, but adequate for single-UVicorn worker. Keys are arbitrary hashables. Values are any JSON-serializable structures. """ def __init__(self, default_ttl: int = 60, maxsize: int = 256) -> None: self.default_ttl = default_ttl self.maxsize = maxsize self._data: Dict[Any, Tuple[float, Any]] = {} def _now(self) -> float: return time.time() def get(self, key: Any) -> Optional[Any]: item = self._data.get(key) if not item: return None expires_at, value = item if expires_at < self._now(): # expired self._data.pop(key, None) return None return value def set(self, key: Any, value: Any, ttl: Optional[int] = None) -> None: if len(self._data) >= self.maxsize: # naive eviction: remove oldest item try: oldest_key = min(self._data.items(), key=lambda kv: kv[1][0])[0] self._data.pop(oldest_key, None) except ValueError: self._data.clear() expires = self._now() + (ttl if ttl is not None else self.default_ttl) self._data[key] = (expires, value) def purge(self) -> None: now = self._now() for k, (exp, _) in list(self._data.items()): if exp < now: self._data.pop(k, None)