Add in-memory read cache with CUD-based invalidation

backend/app/services/runtime_cache.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from threading import RLock
+import time
+from typing import Callable, TypeVar
+
+T = TypeVar("T")
+
+
+@dataclass
+class _CacheEntry:
+    value: object
+    expires_at: float
+    revision: int
+
+
+class RuntimeCache:
+    """Simple in-memory cache for local/prototype use.
+
+    Cache is globally invalidated by `bump_revision()` which we call after CUD.
+    """
+
+    def __init__(self) -> None:
+        self._lock = RLock()
+        self._revision = 0
+        self._entries: dict[str, _CacheEntry] = {}
+
+    def get(self, key: str) -> object | None:
+        now = time.time()
+        with self._lock:
+            entry = self._entries.get(key)
+            if not entry:
+                return None
+            if entry.expires_at <= now or entry.revision != self._revision:
+                self._entries.pop(key, None)
+                return None
+            return entry.value
+
+    def set(self, key: str, value: object, ttl_seconds: int = 30) -> object:
+        now = time.time()
+        with self._lock:
+            self._entries[key] = _CacheEntry(
+                value=value,
+                expires_at=now + max(ttl_seconds, 1),
+                revision=self._revision,
+            )
+            if len(self._entries) > 2000:
+                self._entries.clear()
+        return value
+
+    def get_or_set(self, key: str, factory: Callable[[], T], ttl_seconds: int = 30) -> T:
+        cached = self.get(key)
+        if cached is not None:
+            return cached  # type: ignore[return-value]
+        return self.set(key, factory(), ttl_seconds=ttl_seconds)  # type: ignore[return-value]
+
+    def bump_revision(self) -> int:
+        with self._lock:
+            self._revision += 1
+            if self._revision > 1_000_000_000:
+                self._revision = 1
+            self._entries.clear()
+            return self._revision
+
+    def revision(self) -> int:
+        with self._lock:
+            return self._revision
+
+
+runtime_cache = RuntimeCache()