import threading from collections import OrderedDict from typing import Generic, Hashable, Optional, TypeVar K = TypeVar("K", bound=Hashable) V = TypeVar("V") class LruCache(Generic[K, V]): def __init__(self, capacity: int): self.capacity = max(1, capacity) self._items: "OrderedDict[K, V]" = OrderedDict() self._lock = threading.RLock() def get(self, key: K) -> Optional[V]: with self._lock: value = self._items.get(key) if value is None: return None self._items.move_to_end(key) return value 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 clear(self) -> None: with self._lock: self._items.clear() def __len__(self) -> int: with self._lock: return len(self._items)