aaronrene / overseer-kit public
cache.py python
56 lines 1.6 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago
1 """Ephemeral in-memory cache of fetched bytes + sha256 (§HGD.4.3).
2
3 Never stores secrets. Never authoritative over a fresh upstream response at refresh.
4 """
5
6 from __future__ import annotations
7
8 import hashlib
9 import threading
10 from dataclasses import dataclass
11 from typing import Any
12
13
14 @dataclass(frozen=True)
15 class CacheEntry:
16 """Cached remote bytes with content digest."""
17
18 raw: bytes
19 sha256: str
20 payload: Any
21
22
23 class EphemeralByteCache:
24 """Thread-safe in-memory cache keyed by opaque strings."""
25
26 def __init__(self, *, max_entries: int = 256) -> None:
27 self._max = max_entries
28 self._lock = threading.Lock()
29 self._store: dict[str, CacheEntry] = {}
30
31 @staticmethod
32 def digest(raw: bytes) -> str:
33 """Return lowercase hex sha256 of ``raw``."""
34 return hashlib.sha256(raw).hexdigest()
35
36 def get(self, key: str) -> CacheEntry | None:
37 with self._lock:
38 return self._store.get(key)
39
40 def put(self, key: str, raw: bytes, payload: Any = None) -> CacheEntry:
41 entry = CacheEntry(raw=raw, sha256=self.digest(raw), payload=payload)
42 with self._lock:
43 if len(self._store) >= self._max and key not in self._store:
44 # Drop an arbitrary oldest-ish key (insertion order on 3.7+).
45 oldest = next(iter(self._store))
46 del self._store[oldest]
47 self._store[key] = entry
48 return entry
49
50 def clear(self) -> None:
51 with self._lock:
52 self._store.clear()
53
54 def __len__(self) -> int:
55 with self._lock:
56 return len(self._store)
File History 1 commit
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago