124 lines
4.0 KiB
Python
124 lines
4.0 KiB
Python
"""Thread-safe LRU cache with per-entry TTL.
|
|
|
|
Entries expire after ``ttl`` seconds (measured from insertion/refresh). Access
|
|
to an expired entry treats it as a miss and evicts it. ``maxsize`` bounds the
|
|
number of live entries; LRU eviction removes the least-recently-used entry.
|
|
|
|
Correctness notes:
|
|
- A single lock guards the ordered dict and expiry bookkeeping.
|
|
- ``get`` refreshes recency on hit (move-to-end) but does NOT extend TTL
|
|
(TTL is fixed at put time) -- matches "cache for at most ttl seconds".
|
|
- Expired entries are lazily removed on get/put; a background sweeper is
|
|
intentionally omitted to keep the module dependency-free and deterministic.
|
|
- ``put`` overwrites an existing key, resetting its TTL and recency.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from collections import OrderedDict
|
|
from typing import Any, Callable, Dict, Optional
|
|
|
|
|
|
class LRUTTLCache:
|
|
def __init__(self, maxsize: int = 128, ttl: float = 60.0, clock: Callable[[], float] = time.monotonic) -> None:
|
|
if maxsize < 1:
|
|
raise ValueError("maxsize must be >= 1")
|
|
if ttl <= 0:
|
|
raise ValueError("ttl must be > 0")
|
|
self._maxsize = maxsize
|
|
self._ttl = ttl
|
|
self._clock = clock
|
|
self._lock = threading.Lock()
|
|
# key -> (value, expiry)
|
|
self._data: "OrderedDict[Any, tuple[Any, float]]" = OrderedDict()
|
|
self._hits = 0
|
|
self._misses = 0
|
|
self._evictions = 0
|
|
|
|
@property
|
|
def maxsize(self) -> int:
|
|
return self._maxsize
|
|
|
|
@property
|
|
def ttl(self) -> float:
|
|
return self._ttl
|
|
|
|
def __len__(self) -> int:
|
|
with self._lock:
|
|
self._purge_expired_locked()
|
|
return len(self._data)
|
|
|
|
def _purge_expired_locked(self) -> None:
|
|
now = self._clock()
|
|
# OrderedDict is insertion/recency ordered; expired may be anywhere, scan.
|
|
expired = [k for k, (_, exp) in self._data.items() if exp <= now]
|
|
for k in expired:
|
|
del self._data[k]
|
|
self._evictions += 1
|
|
|
|
def get(self, key: Any, default: Any = None) -> Any:
|
|
with self._lock:
|
|
entry = self._data.get(key)
|
|
if entry is None:
|
|
self._misses += 1
|
|
return default
|
|
value, expiry = entry
|
|
now = self._clock()
|
|
if expiry <= now:
|
|
# expired -> miss + evict
|
|
del self._data[key]
|
|
self._misses += 1
|
|
self._evictions += 1
|
|
return default
|
|
# hit: refresh recency
|
|
self._data.move_to_end(key)
|
|
self._hits += 1
|
|
return value
|
|
|
|
def put(self, key: Any, value: Any) -> None:
|
|
with self._lock:
|
|
self._purge_expired_locked()
|
|
expiry = self._clock() + self._ttl
|
|
if key in self._data:
|
|
self._data[key] = (value, expiry)
|
|
self._data.move_to_end(key)
|
|
return
|
|
# evict LRU if at capacity
|
|
while len(self._data) >= self._maxsize:
|
|
self._data.popitem(last=False)
|
|
self._evictions += 1
|
|
self._data[key] = (value, expiry)
|
|
|
|
def delete(self, key: Any) -> bool:
|
|
with self._lock:
|
|
if key in self._data:
|
|
del self._data[key]
|
|
return True
|
|
return False
|
|
|
|
def clear(self) -> None:
|
|
with self._lock:
|
|
self._data.clear()
|
|
|
|
def contains(self, key: Any) -> bool:
|
|
with self._lock:
|
|
entry = self._data.get(key)
|
|
if entry is None:
|
|
return False
|
|
_, expiry = entry
|
|
if expiry <= self._clock():
|
|
del self._data[key]
|
|
self._evictions += 1
|
|
return False
|
|
return True
|
|
|
|
@property
|
|
def stats(self) -> Dict[str, int]:
|
|
with self._lock:
|
|
self._purge_expired_locked()
|
|
return {"size": len(self._data), "hits": self._hits, "misses": self._misses, "evictions": self._evictions}
|
|
|
|
|
|
# (sentinel removed)
|