| from fastapi_cache import FastAPICache |
| from fastapi_cache.backends.inmemory import InMemoryBackend |
| from fastapi_cache.decorator import cache |
| from datetime import timedelta |
| from app.config import CACHE_TTL |
| import logging |
| from app.utils.logging import LogFormatter |
|
|
| logger = logging.getLogger(__name__) |
|
|
| def setup_cache(): |
| """Initialize FastAPI Cache with in-memory backend""" |
| FastAPICache.init( |
| backend=InMemoryBackend(), |
| prefix="fastapi-cache", |
| expire=CACHE_TTL |
| ) |
| logger.info(LogFormatter.success("FastAPI Cache initialized with in-memory backend")) |
|
|
| def invalidate_cache_key(key: str): |
| """Invalidate a specific cache key""" |
| try: |
| backend = FastAPICache.get_backend() |
| if hasattr(backend, 'delete'): |
| 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(namespace: str, *args) -> str: |
| """Build a consistent cache key""" |
| key = f"fastapi-cache:{namespace}:{':'.join(str(arg) for arg in args)}" |
| logger.debug(LogFormatter.info(f"Built cache key: {key}")) |
| return key |
|
|
| def cached(expire: int = CACHE_TTL, key_builder=None): |
| """Decorator for caching endpoint responses |
| |
| Args: |
| expire (int): Cache TTL in seconds |
| key_builder (callable, optional): Custom key builder function |
| """ |
| return cache( |
| expire=expire, |
| key_builder=key_builder |
| ) |