47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Any, Dict, Tuple, Optional
|
|
|
|
|
|
class TTLCache:
|
|
"""Simple in-memory TTL cache (per-process). Not thread-safe, but adequate for single-UVicorn worker.
|
|
Keys are arbitrary hashables. Values are any JSON-serializable structures.
|
|
"""
|
|
|
|
def __init__(self, default_ttl: int = 60, maxsize: int = 256) -> None:
|
|
self.default_ttl = default_ttl
|
|
self.maxsize = maxsize
|
|
self._data: Dict[Any, Tuple[float, Any]] = {}
|
|
|
|
def _now(self) -> float:
|
|
return time.time()
|
|
|
|
def get(self, key: Any) -> Optional[Any]:
|
|
item = self._data.get(key)
|
|
if not item:
|
|
return None
|
|
expires_at, value = item
|
|
if expires_at < self._now():
|
|
# expired
|
|
self._data.pop(key, None)
|
|
return None
|
|
return value
|
|
|
|
def set(self, key: Any, value: Any, ttl: Optional[int] = None) -> None:
|
|
if len(self._data) >= self.maxsize:
|
|
# naive eviction: remove oldest item
|
|
try:
|
|
oldest_key = min(self._data.items(), key=lambda kv: kv[1][0])[0]
|
|
self._data.pop(oldest_key, None)
|
|
except ValueError:
|
|
self._data.clear()
|
|
expires = self._now() + (ttl if ttl is not None else self.default_ttl)
|
|
self._data[key] = (expires, value)
|
|
|
|
def purge(self) -> None:
|
|
now = self._now()
|
|
for k, (exp, _) in list(self._data.items()):
|
|
if exp < now:
|
|
self._data.pop(k, None)
|