Spaces:
Running
Running
| """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) | |