Spaces:
Running
Running
| 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 | |
| ) | |