beam: LRU cache with per-entry TTL (beam variant) + tests

wizard-work/beam/lru_ttl_cache_beam.py

@@ -0,0 +1,123 @@
+"""Thread-safe LRU cache with per-entry TTL.
+
+Entries expire after ``ttl`` seconds (measured from insertion/refresh). Access
+to an expired entry treats it as a miss and evicts it. ``maxsize`` bounds the
+number of live entries; LRU eviction removes the least-recently-used entry.
+
+Correctness notes:
+- A single lock guards the ordered dict and expiry bookkeeping.
+- ``get`` refreshes recency on hit (move-to-end) but does NOT extend TTL
+  (TTL is fixed at put time) -- matches "cache for at most ttl seconds".
+- Expired entries are lazily removed on get/put; a background sweeper is
+  intentionally omitted to keep the module dependency-free and deterministic.
+- ``put`` overwrites an existing key, resetting its TTL and recency.
+"""
+from __future__ import annotations
+
+import threading
+import time
+from collections import OrderedDict
+from typing import Any, Callable, Dict, Optional
+
+
+class LRUTTLCache:
+    def __init__(self, maxsize: int = 128, ttl: float = 60.0, clock: Callable[[], float] = time.monotonic) -> None:
+        if maxsize < 1:
+            raise ValueError("maxsize must be >= 1")
+        if ttl <= 0:
+            raise ValueError("ttl must be > 0")
+        self._maxsize = maxsize
+        self._ttl = ttl
+        self._clock = clock
+        self._lock = threading.Lock()
+        # key -> (value, expiry)
+        self._data: "OrderedDict[Any, tuple[Any, float]]" = OrderedDict()
+        self._hits = 0
+        self._misses = 0
+        self._evictions = 0
+
+    @property
+    def maxsize(self) -> int:
+        return self._maxsize
+
+    @property
+    def ttl(self) -> float:
+        return self._ttl
+
+    def __len__(self) -> int:
+        with self._lock:
+            self._purge_expired_locked()
+            return len(self._data)
+
+    def _purge_expired_locked(self) -> None:
+        now = self._clock()
+        # OrderedDict is insertion/recency ordered; expired may be anywhere, scan.
+        expired = [k for k, (_, exp) in self._data.items() if exp <= now]
+        for k in expired:
+            del self._data[k]
+            self._evictions += 1
+
+    def get(self, key: Any, default: Any = None) -> Any:
+        with self._lock:
+            entry = self._data.get(key)
+            if entry is None:
+                self._misses += 1
+                return default
+            value, expiry = entry
+            now = self._clock()
+            if expiry <= now:
+                # expired -> miss + evict
+                del self._data[key]
+                self._misses += 1
+                self._evictions += 1
+                return default
+            # hit: refresh recency
+            self._data.move_to_end(key)
+            self._hits += 1
+            return value
+
+    def put(self, key: Any, value: Any) -> None:
+        with self._lock:
+            self._purge_expired_locked()
+            expiry = self._clock() + self._ttl
+            if key in self._data:
+                self._data[key] = (value, expiry)
+                self._data.move_to_end(key)
+                return
+            # evict LRU if at capacity
+            while len(self._data) >= self._maxsize:
+                self._data.popitem(last=False)
+                self._evictions += 1
+            self._data[key] = (value, expiry)
+
+    def delete(self, key: Any) -> bool:
+        with self._lock:
+            if key in self._data:
+                del self._data[key]
+                return True
+            return False
+
+    def clear(self) -> None:
+        with self._lock:
+            self._data.clear()
+
+    def contains(self, key: Any) -> bool:
+        with self._lock:
+            entry = self._data.get(key)
+            if entry is None:
+                return False
+            _, expiry = entry
+            if expiry <= self._clock():
+                del self._data[key]
+                self._evictions += 1
+                return False
+            return True
+
+    @property
+    def stats(self) -> Dict[str, int]:
+        with self._lock:
+            self._purge_expired_locked()
+            return {"size": len(self._data), "hits": self._hits, "misses": self._misses, "evictions": self._evictions}
+
+
+# (sentinel removed)

wizard-work/beam/test_lru_ttl_cache_beam.py

@@ -0,0 +1,78 @@
+import threading
+import time
+import unittest
+
+from lru_ttl_cache_beam import LRUTTLCache
+
+
+class TestLRUTTLCache(unittest.TestCase):
+    def test_basic_put_get(self):
+        c = LRUTTLCache(maxsize=2, ttl=10)
+        c.put("a", 1)
+        self.assertEqual(c.get("a"), 1)
+        self.assertEqual(c.get("missing", "d"), "d")
+
+    def test_lru_eviction(self):
+        c = LRUTTLCache(maxsize=2, ttl=10)
+        c.put("a", 1); c.put("b", 2)
+        c.get("a")  # refresh a
+        c.put("c", 3)  # evicts b
+        self.assertEqual(c.get("b", None), None)
+        self.assertEqual(c.get("a"), 1)
+        self.assertEqual(c.get("c"), 3)
+
+    def test_ttl_expiry(self):
+        clock = {"t": 0.0}
+        c = LRUTTLCache(maxsize=4, ttl=5, clock=lambda: clock["t"])
+        c.put("a", 1)
+        self.assertTrue(c.contains("a"))
+        clock["t"] = 6.0
+        self.assertFalse(c.contains("a"))
+        self.assertEqual(c.get("a", "x"), "x")
+
+    def test_put_overwrite_resets_ttl(self):
+        clock = {"t": 0.0}
+        c = LRUTTLCache(maxsize=4, ttl=5, clock=lambda: clock["t"])
+        c.put("a", 1)
+        clock["t"] = 4.0
+        c.put("a", 2)  # reset ttl: now expires at 9
+        clock["t"] = 8.0
+        self.assertEqual(c.get("a"), 2)
+        clock["t"] = 10.0
+        self.assertEqual(c.get("a", "x"), "x")
+
+    def test_stats(self):
+        c = LRUTTLCache(maxsize=2, ttl=10)
+        c.put("a", 1); c.put("b", 2)
+        c.get("a"); c.get("zzz")
+        s = c.stats
+        self.assertEqual(s["hits"], 1)
+        self.assertEqual(s["misses"], 1)
+        self.assertEqual(s["size"], 2)
+
+    def test_thread_safety(self):
+        c = LRUTTLCache(maxsize=100, ttl=100)
+        errs = []
+
+        def w(i):
+            try:
+                for j in range(50):
+                    c.put((i, j), i * j)
+                    c.get((i, j))
+            except Exception as e:  # noqa
+                errs.append(e)
+
+        ts = [threading.Thread(target=w, args=(i,)) for i in range(8)]
+        for t in ts: t.start()
+        for t in ts: t.join()
+        self.assertFalse(errs)
+
+    def test_validation(self):
+        with self.assertRaises(ValueError):
+            LRUTTLCache(maxsize=0, ttl=1)
+        with self.assertRaises(ValueError):
+            LRUTTLCache(maxsize=1, ttl=0)
+
+
+if __name__ == "__main__":
+    unittest.main()