"""Small per-instance LRU caches; no global references to fitted codecs.""" from collections import OrderedDict from collections.abc import MutableMapping from threading import RLock class BoundedCache(MutableMapping): """Evict least-recently-used entries above a fixed capacity. Args: capacity: Maximum retained entries (positive integer). """ def __init__(self, capacity): if type(capacity) is not int or capacity < 1: raise ValueError("cache capacity must be a positive integer") self.capacity = capacity self._items = OrderedDict() self._lock = RLock() def __getitem__(self, key): with self._lock: value = self._items[key] self._items.move_to_end(key) return value def __setitem__(self, key, value): 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 __delitem__(self, key): with self._lock: del self._items[key] def __iter__(self): with self._lock: return iter(tuple(self._items)) def __len__(self): with self._lock: return len(self._items)