Spaces:
Sleeping
Sleeping
File size: 2,929 Bytes
db284b9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | """A tiny bounded LRU (Least-Recently-Used) cache backed by OrderedDict.
Why not functools.lru_cache? These caches are SHARED between two endpoints (the
blocking and the streaming answer handlers read *and* write the same store),
they need an explicit `cached` flag in the HTTP response, and they must support
a per-graph .clear() when the active graph is hot-swapped. A decorator gives you
none of those. ~25 lines of OrderedDict buys all of it with O(1) get/set/evict
and a hard memory ceiling, which is the whole point — the old plain-dict caches
grew without bound and leaked memory on a long-running server.
"""
from __future__ import annotations
from collections import OrderedDict
from typing import Generic, Optional, TypeVar
K = TypeVar("K")
V = TypeVar("V")
class LRUCache(Generic[K, V]):
"""Fixed-capacity cache that evicts the least-recently-used entry first.
The eviction policy IS the OrderedDict's ordering: we keep the *oldest*
(least-recently-used) entry on the left and the *most-recently-used* on the
right. Every read or write moves the touched key to the right, so whatever
has drifted to the far left is, by definition, the coldest entry — and that's
exactly what we drop when we're over capacity.
"""
def __init__(self, capacity: int = 256) -> None:
if capacity <= 0:
raise ValueError("LRUCache capacity must be positive")
self.capacity = capacity
self._store: "OrderedDict[K, V]" = OrderedDict()
def get(self, key: K) -> Optional[V]:
"""Return the value (and mark it most-recently-used), or None on miss."""
if key not in self._store:
return None
# A successful read counts as "use": promote the key to the right end so
# it survives the next eviction. This is what makes it LRU and not FIFO.
self._store.move_to_end(key)
return self._store[key]
def __contains__(self, key: K) -> bool:
# Note: membership does NOT promote — only get()/set() count as a "use".
return key in self._store
def set(self, key: K, value: V) -> None:
"""Insert/overwrite a value, then evict oldest entries past capacity."""
if key in self._store:
# Overwriting an existing key still counts as a use → promote it.
self._store.move_to_end(key)
self._store[key] = value
# Evict from the LEFT (oldest) until we're back within the ceiling.
# last=False pops the first-inserted / least-recently-used item; a normal
# popitem() would pop the newest, which is the opposite of what we want.
while len(self._store) > self.capacity:
self._store.popitem(last=False)
def clear(self) -> None:
"""Drop everything — used when the active graph changes (snippets differ)."""
self._store.clear()
def __len__(self) -> int:
return len(self._store)
|