Spaces:
Running
Running
| from datetime import timedelta | |
| from app.config import CACHE_TTL | |
| import logging | |
| from app.core.formatting import LogFormatter | |
| from typing import Optional, Any | |
| from functools import wraps | |
| logger = logging.getLogger(__name__) | |
| # Try to import fastapi-cache2; fall back to no-op if unavailable | |
| try: | |
| from fastapi_cache import FastAPICache | |
| from fastapi_cache.backends.inmemory import InMemoryBackend | |
| from fastapi_cache.decorator import cache as _cache_decorator | |
| CACHE_AVAILABLE = True | |
| except ImportError: | |
| logger.warning("fastapi-cache2 not available; caching disabled") | |
| CACHE_AVAILABLE = False | |
| if CACHE_AVAILABLE: | |
| class CustomInMemoryBackend(InMemoryBackend): | |
| def __init__(self): | |
| """Initialize the cache backend""" | |
| super().__init__() | |
| self._store = {} | |
| async def delete(self, key: str) -> bool: | |
| """Delete a key from the cache""" | |
| try: | |
| if key in self._store: | |
| del self._store[key] | |
| return True | |
| return False | |
| except Exception as e: | |
| logger.error(LogFormatter.error(f"Failed to delete key {key} from cache", e)) | |
| return False | |
| async def get(self, key: str) -> Any: | |
| """Get a value from the cache""" | |
| return self._store.get(key) | |
| async def set(self, key: str, value: Any, expire: Optional[int] = None) -> None: | |
| """Set a value in the cache""" | |
| self._store[key] = value | |
| def setup_cache(): | |
| """Initialize FastAPI Cache with in-memory backend""" | |
| if not CACHE_AVAILABLE: | |
| logger.warning(LogFormatter.warning("Cache setup skipped — fastapi-cache2 not installed")) | |
| return | |
| try: | |
| logger.info(LogFormatter.section("CACHE INITIALIZATION")) | |
| FastAPICache.init( | |
| backend=CustomInMemoryBackend(), | |
| prefix="fastapi-cache" | |
| ) | |
| logger.info(LogFormatter.success("Cache initialized successfully")) | |
| except Exception as e: | |
| logger.error(LogFormatter.error("Failed to initialize cache", e)) | |
| # Don't raise — let the app run without cache | |
| async def invalidate_cache_key(key: str): | |
| """Invalidate a specific cache key""" | |
| if not CACHE_AVAILABLE: | |
| return | |
| try: | |
| backend = FastAPICache.get_backend() | |
| if hasattr(backend, 'delete'): | |
| await backend.delete(key) | |
| logger.info(LogFormatter.success(f"Cache invalidated for key: {key}")) | |
| else: | |
| logger.warning(LogFormatter.warning("Cache backend does not support deletion")) | |
| except Exception as e: | |
| logger.error(LogFormatter.error(f"Failed to invalidate cache key: {key}", e)) | |
| def build_cache_key(*args) -> str: | |
| """Build a cache key from multiple arguments""" | |
| return ":".join(str(arg) for arg in args if arg is not None) | |
| def cached(expire: int = CACHE_TTL, key_builder=None): | |
| """Decorator for caching endpoint responses. | |
| Falls back to a no-op decorator when fastapi-cache2 is not installed. | |
| """ | |
| if CACHE_AVAILABLE: | |
| return _cache_decorator( | |
| expire=expire, | |
| key_builder=key_builder | |
| ) | |
| else: | |
| # No-op decorator | |
| def decorator(func): | |
| async def wrapper(*args, **kwargs): | |
| return await func(*args, **kwargs) | |
| return wrapper | |
| return decorator | |