#419: basic cache class

Author: Kraysent

app/lib/cache/__init__.py

@@ -0,0 +1,69 @@
+import concurrent.futures
+import threading
+from collections.abc import Callable
+from datetime import timedelta
+from time import monotonic
+
+import pydantic
+import structlog
+
+
+class BackgroundCache[T: pydantic.BaseModel]:
+    def __init__(
+        self,
+        name: str,
+        refresh_func: Callable[[], T],
+        refresh_frequency: timedelta,
+        refresh_timeout: timedelta,
+    ) -> None:
+        self.name = name
+        self._refresh_func = refresh_func
+        self.refresh_frequency = refresh_frequency
+        self._refresh_timeout = refresh_timeout
+        self._lock = threading.Lock()
+        self._stop_event = threading.Event()
+        self._value: T = self._do_refresh()
+
+    def _do_refresh(self) -> T:
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+            future = executor.submit(self._refresh_func)
+            try:
+                return future.result(timeout=self._refresh_timeout.total_seconds())
+            except concurrent.futures.TimeoutError as e:
+                raise TimeoutError(f"refresh_func timed out after {self._refresh_timeout}") from e
+
+    def get(self) -> T:
+        with self._lock:
+            return self._value
+
+    def run(self) -> None:
+        log = structlog.get_logger().bind(cache_name=self.name)
+
+        while not self._stop_event.is_set():
+            start = monotonic()
+
+            try:
+                new_value = self._do_refresh()
+            except Exception as e:
+                log.error("cache refresh failed", reason=str(e), exc_info=True)
+                self._stop_event.wait(timeout=self.refresh_frequency.total_seconds())
+                continue
+
+            elapsed = monotonic() - start
+            with self._lock:
+                old_value = self._value
+                self._value = new_value
+
+            old_json = old_value.model_dump_json()
+            new_json = new_value.model_dump_json()
+            changed = old_json != new_json
+            log.debug(
+                "cache refreshed",
+                duration_seconds=round(elapsed, 3),
+                size_bytes=len(new_json),
+                changed=changed,
+            )
+            self._stop_event.wait(timeout=self.refresh_frequency.total_seconds())
+
+    def stop(self) -> None:
+        self._stop_event.set()

app/lib/cache/cache.py

@@ -0,0 +1,99 @@
+import threading
+import time
+import unittest
+from datetime import timedelta
+
+import pydantic
+
+from app.lib.cache.cache import BackgroundCache
+
+
+class DummyModel(pydantic.BaseModel):
+    value: int
+
+
+class BackgroundCacheTest(unittest.TestCase):
+    def test_happy_path_no_change(self):
+        value = DummyModel(value=1)
+
+        def refresh():
+            return value
+
+        cache = BackgroundCache(
+            name="test_cache",
+            refresh_func=refresh,
+            refresh_frequency=timedelta(seconds=0.1),
+            refresh_timeout=timedelta(seconds=1),
+        )
+        self.assertEqual(cache.get().value, 1)
+        t = threading.Thread(target=cache.run, daemon=True)
+        t.start()
+        time.sleep(0.35)
+        for _ in range(3):
+            self.assertEqual(cache.get().value, 1)
+        cache.stop()
+        t.join(timeout=1)
+        self.assertEqual(cache.get().value, 1)
+
+    def test_happy_path_with_change(self):
+        state = {"v": 0}
+
+        def refresh():
+            return DummyModel(value=state["v"])
+
+        cache = BackgroundCache(
+            name="test_cache",
+            refresh_func=refresh,
+            refresh_frequency=timedelta(seconds=0.05),
+            refresh_timeout=timedelta(seconds=1),
+        )
+        self.assertEqual(cache.get().value, 0)
+        t = threading.Thread(target=cache.run, daemon=True)
+        t.start()
+        self.assertEqual(cache.get().value, 0)
+        state["v"] = 1
+        time.sleep(0.15)
+        self.assertEqual(cache.get().value, 1)
+        state["v"] = 2
+        time.sleep(0.15)
+        self.assertEqual(cache.get().value, 2)
+        cache.stop()
+        t.join(timeout=1)
+
+    def test_refresh_fails_then_restores_value_unchanged(self):
+        call_count = 0
+
+        def refresh():
+            nonlocal call_count
+            call_count += 1
+            if call_count == 2:
+                raise RuntimeError("transient failure")
+            return DummyModel(value=42)
+
+        cache = BackgroundCache(
+            name="test_cache",
+            refresh_func=refresh,
+            refresh_frequency=timedelta(seconds=0.05),
+            refresh_timeout=timedelta(seconds=1),
+        )
+        self.assertEqual(cache.get().value, 42)
+        t = threading.Thread(target=cache.run, daemon=True)
+        t.start()
+        time.sleep(0.2)
+        self.assertEqual(cache.get().value, 42)
+        cache.stop()
+        t.join(timeout=1)
+        self.assertEqual(cache.get().value, 42)
+
+    def test_refresh_fails_during_init_raises(self):
+        def refresh():
+            raise ValueError("init failed")
+
+        with self.assertRaises(ValueError) as ctx:
+            BackgroundCache(
+                name="test_cache",
+                refresh_func=refresh,
+                refresh_frequency=timedelta(seconds=1),
+                refresh_timeout=timedelta(seconds=1),
+            )
+        self.assertIn("init failed", str(ctx.exception))