Spaces:
Running
Running
File size: 1,486 Bytes
707993f | 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 | from __future__ import annotations
import threading
from typing import Any
from cachetools import TTLCache
from config import settings
class SimpleCache:
def __init__(self, maxsize: int, ttl: int, enabled: bool = True) -> None:
self._enabled = enabled
self._lock = threading.Lock()
self._store: TTLCache | None = TTLCache(maxsize=maxsize, ttl=ttl) if enabled else None
def get(self, key: str) -> Any | None:
if not self._enabled:
return None
with self._lock:
return self._store.get(key)
def set(self, key: str, value: Any) -> None:
if not self._enabled:
return
with self._lock:
self._store[key] = value
def clear(self) -> None:
if self._enabled:
with self._lock:
self._store.clear()
def stats(self) -> dict:
if not self._enabled:
return {"enabled": False, "size": 0, "maxsize": 0}
with self._lock:
return {
"enabled": True,
"size": len(self._store),
"maxsize": self._store.maxsize,
}
def make_cache_key(*parts: str) -> str:
return "||".join(parts)
embedding_cache = SimpleCache(
maxsize=settings.cache_max_size, ttl=settings.cache_ttl_seconds, enabled=settings.enable_cache
)
answer_cache = SimpleCache(
maxsize=settings.cache_max_size, ttl=settings.cache_ttl_seconds, enabled=settings.enable_cache
)
|