142 lines
4.3 KiB
Python
142 lines
4.3 KiB
Python
"""TTL-aware LRU cache.
|
|
|
|
A small, dependency-free, thread-safe-ish (single-threaded design) cache
|
|
combining least-recently-used eviction with per-entry time-to-live expiry.
|
|
|
|
Design notes:
|
|
- Ordering is tracked with an OrderedDict (move_to_end on access).
|
|
- Expiry is checked lazily on get/peek; expired entries are dropped on access.
|
|
- Capacity counts only live (non-expired) entries at insertion time; expired
|
|
entries are purged opportunistically when capacity is exceeded.
|
|
- A monotonic clock (time.monotonic) is used so wall-clock jumps don't break TTL.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import OrderedDict
|
|
from dataclasses import dataclass
|
|
from typing import Any, Callable, Generic, Optional, TypeVar
|
|
|
|
K = TypeVar("K")
|
|
V = TypeVar("V")
|
|
|
|
|
|
class CacheMiss(KeyError):
|
|
"""Raised by get_strict when a key is absent or expired."""
|
|
|
|
|
|
@dataclass
|
|
class _Entry(Generic[V]):
|
|
value: V
|
|
expires_at: float # monotonic seconds; inf means no expiry
|
|
|
|
|
|
class TTLCache(Generic[K, V]):
|
|
def __init__(
|
|
self,
|
|
capacity: int,
|
|
default_ttl: Optional[float] = None,
|
|
clock: Callable[[], float] = time.monotonic,
|
|
) -> None:
|
|
if capacity <= 0:
|
|
raise ValueError("capacity must be positive")
|
|
if default_ttl is not None and default_ttl <= 0:
|
|
raise ValueError("default_ttl must be positive or None")
|
|
self._capacity = capacity
|
|
self._default_ttl = default_ttl
|
|
self._clock = clock
|
|
self._data: "OrderedDict[K, _Entry[V]]" = OrderedDict()
|
|
|
|
# -- internals ---------------------------------------------------------
|
|
|
|
def _now(self) -> float:
|
|
return self._clock()
|
|
|
|
def _purge_expired(self) -> None:
|
|
# Drop expired entries from the front/back opportunistically.
|
|
now = self._now()
|
|
dead = [k for k, e in self._data.items() if e.expires_at <= now]
|
|
for k in dead:
|
|
del self._data[k]
|
|
|
|
def _ttl_to_deadline(self, ttl: Optional[float]) -> float:
|
|
if ttl is None:
|
|
ttl = self._default_ttl
|
|
if ttl is None:
|
|
return float("inf")
|
|
if ttl <= 0:
|
|
raise ValueError("ttl must be positive")
|
|
return self._now() + ttl
|
|
|
|
# -- public API --------------------------------------------------------
|
|
|
|
def __len__(self) -> int:
|
|
self._purge_expired()
|
|
return len(self._data)
|
|
|
|
def __contains__(self, key: K) -> bool:
|
|
entry = self._data.get(key)
|
|
if entry is None:
|
|
return False
|
|
if entry.expires_at <= self._now():
|
|
del self._data[key]
|
|
return False
|
|
return True
|
|
|
|
def get(self, key: K, default: Any = None) -> Any:
|
|
entry = self._data.get(key)
|
|
if entry is None:
|
|
return default
|
|
if entry.expires_at <= self._now():
|
|
del self._data[key]
|
|
return default
|
|
self._data.move_to_end(key)
|
|
return entry.value
|
|
|
|
def get_strict(self, key: K) -> V:
|
|
v = self.get(key, _MISSING)
|
|
if v is _MISSING:
|
|
raise CacheMiss(key)
|
|
return v # type: ignore[return-value]
|
|
|
|
def set(self, key: K, value: V, ttl: Optional[float] = None) -> None:
|
|
deadline = self._ttl_to_deadline(ttl)
|
|
if key in self._data:
|
|
del self._data[key]
|
|
self._data[key] = _Entry(value=value, expires_at=deadline)
|
|
self._data.move_to_end(key)
|
|
self._evict_if_needed()
|
|
|
|
def _evict_if_needed(self) -> None:
|
|
self._purge_expired()
|
|
while len(self._data) > self._capacity:
|
|
self._data.popitem(last=False)
|
|
|
|
def delete(self, key: K) -> bool:
|
|
if key in self._data:
|
|
del self._data[key]
|
|
return True
|
|
return False
|
|
|
|
def clear(self) -> None:
|
|
self._data.clear()
|
|
|
|
def keys(self):
|
|
self._purge_expired()
|
|
return list(self._data.keys())
|
|
|
|
def items(self):
|
|
self._purge_expired()
|
|
return [(k, e.value) for k, e in self._data.items()]
|
|
|
|
def peek(self, key: K, default: Any = None) -> Any:
|
|
"""Return value without updating recency or triggering purge of others."""
|
|
entry = self._data.get(key)
|
|
if entry is None or entry.expires_at <= self._now():
|
|
return default
|
|
return entry.value
|
|
|
|
|
|
_MISSING = object()
|