Mapping which maintains a maximum size by removing the least recently used value. Items are passed to dispose before being removed and setting an item which is already in the cache has to be done using the replace method.
| 7 | |
| 8 | |
| 9 | class LRUCache(MutableMapping[K, V]): |
| 10 | """ |
| 11 | Mapping which maintains a maximum size by removing the least recently used value. |
| 12 | Items are passed to dispose before being removed and setting an item which is |
| 13 | already in the cache has to be done using the replace method. |
| 14 | """ |
| 15 | |
| 16 | _cache: OrderedDict[K, V] |
| 17 | |
| 18 | _capacity: int |
| 19 | |
| 20 | _dispose: Callable[[V], None] |
| 21 | |
| 22 | def __init__(self, capacity: int, dispose: Callable[[V], None] = lambda _: None): |
| 23 | self._cache = OrderedDict() |
| 24 | self._capacity = capacity |
| 25 | self._dispose = dispose |
| 26 | |
| 27 | def __setitem__(self, key: K, value: V) -> None: |
| 28 | assert ( |
| 29 | key not in self._cache |
| 30 | ), "Unexpected attempt to replace a cached item without first deleting the old item." |
| 31 | while len(self._cache) >= self._capacity: |
| 32 | self._dispose(self._cache.popitem(last=False)[1]) |
| 33 | self._cache[key] = value # add new entry at the end |
| 34 | |
| 35 | def __getitem__(self, key: K) -> V: |
| 36 | self._cache.move_to_end(key) # raise KeyError if not found |
| 37 | return self._cache[key] |
| 38 | |
| 39 | def __delitem__(self, key: K) -> None: |
| 40 | self._dispose(self._cache.pop(key)) |
| 41 | |
| 42 | def __contains__(self, key: object) -> bool: |
| 43 | return key in self._cache |
| 44 | |
| 45 | def __len__(self) -> int: |
| 46 | return len(self._cache) |
| 47 | |
| 48 | def replace(self, key: K, value: V) -> None: |
| 49 | """Replace an item that is already present, not disposing it in the process.""" |
| 50 | # this method complements __setitem__ which should be used for the normal use case. |
| 51 | assert key in self._cache, "Unexpected attempt to update a non-existing item." |
| 52 | self._cache[key] = value |
| 53 | |
| 54 | def clear(self) -> None: |
| 55 | for value in self._cache.values(): |
| 56 | self._dispose(value) |
| 57 | self._cache.clear() |
| 58 | |
| 59 | def __iter__(self) -> Iterator[K]: |
| 60 | return iter(self._cache) |
| 61 | |
| 62 | def keys(self) -> KeysView[K]: |
| 63 | return self._cache.keys() |
| 64 | |
| 65 | def values(self) -> ValuesView[V]: |
| 66 | return self._cache.values() |
no outgoing calls