|
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 |
|
) |