LibBee / src /services /cache_service.py
nikeshn's picture
Upload 6 files
37ae25d verified
Raw
History Blame Contribute Delete
1.31 kB
import time
from typing import Any, Dict, Optional
class CacheService:
def __init__(self, max_size: int = 500):
self._cache: Dict[str, tuple[Any, float]] = {}
self.max_size = max_size
def _prune(self) -> None:
now = time.time()
expired = [k for k, (_, expiry) in self._cache.items() if expiry <= now]
for key in expired:
self._cache.pop(key, None)
if len(self._cache) > self.max_size:
ordered = sorted(self._cache.items(), key=lambda item: item[1][1])
for key, _ in ordered[: max(0, len(self._cache) - self.max_size)]:
self._cache.pop(key, None)
def get(self, key: str) -> Optional[Any]:
item = self._cache.get(key)
if not item:
return None
value, expiry = item
if time.time() >= expiry:
self._cache.pop(key, None)
return None
return value
def set(self, key: str, value: Any, ttl: int = 3600) -> None:
self._prune()
self._cache[key] = (value, time.time() + ttl)
def clear(self) -> None:
self._cache.clear()
def get_stats(self) -> dict:
self._prune()
return {"size": len(self._cache), "keys": list(self._cache.keys())[:10]}