| import json |
| import logging |
| import time |
| from typing import Optional, Any |
| from app.core.config import settings |
|
|
| logger = logging.getLogger("Cache") |
|
|
|
|
| class InMemoryCache: |
| """Simple dict-based cache with TTL support (fallback when Redis is unavailable).""" |
|
|
| def __init__(self): |
| self._data: dict = {} |
| self._expiry: dict = {} |
|
|
| def get(self, key: str) -> Optional[str]: |
| self._cleanup(key) |
| return self._data.get(key) |
|
|
| def setex(self, key: str, ttl: int, value: str): |
| self._data[key] = value |
| self._expiry[key] = time.time() + ttl |
|
|
| def delete(self, *keys): |
| for k in keys: |
| self._data.pop(k, None) |
| self._expiry.pop(k, None) |
|
|
| def keys(self, pattern: str): |
| import fnmatch |
| self._cleanup_all() |
| return [k for k in self._data.keys() if fnmatch.fnmatch(k, pattern)] |
|
|
| def _cleanup(self, key: str): |
| if key in self._expiry and time.time() > self._expiry[key]: |
| self._data.pop(key, None) |
| self._expiry.pop(key, None) |
|
|
| def _cleanup_all(self): |
| now = time.time() |
| expired = [k for k, exp in self._expiry.items() if now > exp] |
| for k in expired: |
| self._data.pop(k, None) |
| self._expiry.pop(k, None) |
|
|
|
|
| class RedisCacheManager: |
| """ |
| 负责题库数据、IRT 参数及计算中间态的缓存加速。 |
| Uses Redis when available, falls back to in-memory cache otherwise. |
| """ |
|
|
| def __init__(self): |
| self.client = None |
| if settings.REDIS_URL: |
| try: |
| import redis |
| self.client = redis.Redis.from_url(settings.REDIS_URL, decode_responses=True) |
| self.client.ping() |
| logger.info("RedisCacheManager connected to Redis") |
| except Exception as e: |
| logger.warning(f"Redis unavailable, using in-memory cache: {e}") |
| self.client = None |
| if self.client is None: |
| self.client = InMemoryCache() |
|
|
| def get_json(self, key: str) -> Optional[Any]: |
| data = self.client.get(key) |
| if data: |
| try: |
| return json.loads(data) |
| except (json.JSONDecodeError, TypeError): |
| return None |
| return None |
|
|
| def set_json(self, key: str, value: Any, ttl_seconds: int = 3600): |
| self.client.setex(key, ttl_seconds, json.dumps(value)) |
|
|
| def delete(self, key: str): |
| self.client.delete(key) |
|
|
| def invalidate_pattern(self, pattern: str): |
| keys = self.client.keys(pattern) |
| if keys: |
| self.client.delete(*keys) |
|
|
|
|
| cache_manager = RedisCacheManager() |
|
|
