File size: 2,723 Bytes
aac350d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | """
In-memory TTL cache.
Keys are strings (typically image hashes + provider name + capability).
Values are arbitrary JSON-serializable objects.
The cache tracks hit/miss counters which the metrics subsystem reads.
"""
from __future__ import annotations
import threading
import time
from collections import OrderedDict
from typing import Any, Optional
class Cache:
"""LRU + TTL cache. Thread-safe."""
def __init__(self, ttl_seconds: int = 3600, max_entries: int = 1000) -> None:
self._ttl = ttl_seconds
self._max = max_entries
self._data: OrderedDict[str, tuple[float, Any]] = OrderedDict()
self._lock = threading.RLock()
self._hits = 0
self._misses = 0
self._evictions = 0
# ------------------------------------------------------------------ #
# Core ops
# ------------------------------------------------------------------ #
def get(self, key: str) -> Optional[Any]:
with self._lock:
entry = self._data.get(key)
if entry is None:
self._misses += 1
return None
expires_at, value = entry
if time.time() > expires_at:
self._data.pop(key, None)
self._misses += 1
return None
self._data.move_to_end(key)
self._hits += 1
return value
def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
with self._lock:
expires_at = time.time() + (ttl if ttl is not None else self._ttl)
self._data[key] = (expires_at, value)
self._data.move_to_end(key)
while len(self._data) > self._max:
self._data.popitem(last=False)
self._evictions += 1
def invalidate(self, key: str) -> bool:
with self._lock:
return self._data.pop(key, None) is not None
def clear(self) -> int:
with self._lock:
n = len(self._data)
self._data.clear()
return n
# ------------------------------------------------------------------ #
# Stats (consumed by metrics + /cache endpoint)
# ------------------------------------------------------------------ #
def stats(self) -> dict:
with self._lock:
total = self._hits + self._misses
return {
"entries": len(self._data),
"max_entries": self._max,
"ttl_seconds": self._ttl,
"hits": self._hits,
"misses": self._misses,
"hit_ratio": round(self._hits / total, 4) if total else 0.0,
"evictions": self._evictions,
}
|