LRU Cache with Size and Hit Stats
Build a thread-safe LRU cache on OrderedDict with capacity eviction, hit and miss counters, and a memoize decorator for pure functions.
Code
import threading
from collections import OrderedDict
from collections.abc import Callable, Hashable
from functools import wraps
from typing import Generic, TypeVar
K = TypeVar("K", bound=Hashable)
V = TypeVar("V")
class LRUCache(Generic[K, V]):
def __init__(self, capacity: int) -> None:
if capacity < 1:
raise ValueError("capacity must be at least 1")
self.capacity = capacity
self._items: OrderedDict[K, V] = OrderedDict()
self._lock = threading.RLock()
self.hits = 0
self.misses = 0
def get(self, key: K) -> V | None:
with self._lock:
if key not in self._items:
self.misses += 1
return None
self.hits += 1
self._items.move_to_end(key)
return self._items[key]
def put(self, key: K, value: V) -> None:
with self._lock:
self._items[key] = value
self._items.move_to_end(key)
while len(self._items) > self.capacity:
self._items.popitem(last=False)
def stats(self) -> dict[str, int | float]:
total = self.hits + self.misses
return {
"size": len(self._items),
"hits": self.hits,
"misses": self.misses,
"hit_rate": round(self.hits / total, 3) if total else 0.0,
}
def memoize(
cache: LRUCache[K, V],
) -> Callable[[Callable[[K], V]], Callable[[K], V]]:
def decorator(func: Callable[[K], V]) -> Callable[[K], V]:
@wraps(func)
def wrapper(key: K) -> V:
cached = cache.get(key)
if cached is not None:
return cached
value = func(key)
cache.put(key, value)
return value
return wrapper
return decorator
def main() -> None:
cache: LRUCache[str, int] = LRUCache(capacity=2)
calls = 0
@memoize(cache)
def slow_square(text: str) -> int:
nonlocal calls
calls += 1
return int(text) ** 2
print(slow_square("4"), slow_square("4"))
print(slow_square("5"), slow_square("6"))
print(slow_square("4"))
print(cache.stats(), "calls:", calls)
if __name__ == "__main__":
main()
Report a problem with this exampleOpens a pre-filled issue on GitHub.