"""In-process, generic TTL cache. Small, typed, and deliberately boring. Used by :mod:`app.services.posts` and :mod:`app.services.pages` to sit in front of the hottest queries (published-posts list, page-by-slug); a 60 s default TTL keeps the site's three-digit daily requests out of the SQLite query path without any cross-process coordination. Not thread-safe in the strict sense — Python's GIL makes the dict operations atomic at CPython bytecode granularity, and worst case a concurrent writer causes a benign duplicate DB read. That is acceptable at this scale; if the site ever grows teeth we can revisit. """ from __future__ import annotations import time from typing import Generic, Hashable, Optional, TypeVar # TypeVar bound to ``Hashable`` so callers cannot accidentally key by a # mutable collection (which would later look up with a different hash # after mutation and silently miss the cache). K = TypeVar("K", bound=Hashable) V = TypeVar("V") class TTLCache(Generic[K, V]): """Tiny TTL-based dict-style cache. Entries expire ``ttl_seconds`` after insertion. Expired entries are dropped lazily on access — there is no background sweep, and the cache is not bounded in size. For our workload (at most a few dozen keys per instance) this is fine. Two operations are public: - :meth:`get` returns the cached value or ``None``. - :meth:`set` stores a value with an expiry. - :meth:`invalidate_all` clears every entry; used by admin-write paths in Phase 4. """ def __init__(self, ttl_seconds: float = 60.0) -> None: """Construct an empty cache. Parameters ---------- ttl_seconds: Time-to-live for every entry, in seconds. 60 s matches the "Caching Strategy" section of ``docs/ROADMAP.md``. """ if ttl_seconds <= 0: # Defensive: a zero/negative TTL would mean every write # instantly expires, which almost always indicates a bug. raise ValueError("ttl_seconds must be positive") self._ttl: float = float(ttl_seconds) # Stored as (expiry_monotonic_ts, value). Using # ``time.monotonic`` avoids issues if the wall clock jumps. self._store: dict[K, tuple[float, V]] = {} def get(self, key: K) -> Optional[V]: """Return the cached value for ``key`` or ``None`` if absent/expired. Expired entries are deleted as a side effect of the lookup so the store doesn't grow unboundedly with stale data in long-running processes. """ entry = self._store.get(key) if entry is None: return None expiry, value = entry if time.monotonic() >= expiry: # Expired — drop lazily and report miss. self._store.pop(key, None) return None return value def set(self, key: K, value: V) -> None: """Store ``value`` under ``key`` with the configured TTL.""" self._store[key] = (time.monotonic() + self._ttl, value) def invalidate_all(self) -> None: """Drop every cached entry. Called by the Phase 4 admin write path so readers see the new content on the very next request, not up to 60 s later. """ self._store.clear()