File size: 1,660 Bytes
e7abd9e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 |
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
) |