File size: 2,599 Bytes
e7abd9e 23c96f8 5c2e213 e7abd9e 5c2e213 e7abd9e 5c2e213 e7abd9e 5c2e213 e7abd9e 5c2e213 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 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 |
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.core.formatting import LogFormatter
from typing import Optional, Any
logger = logging.getLogger(__name__)
class CustomInMemoryBackend(InMemoryBackend):
def __init__(self):
"""Initialize the cache backend"""
super().__init__()
self.cache = {}
async def delete(self, key: str) -> bool:
"""Delete a key from the cache"""
try:
if key in self.cache:
del self.cache[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.cache.get(key)
async def set(self, key: str, value: Any, expire: Optional[int] = None) -> None:
"""Set a value in the cache"""
self.cache[key] = value
def setup_cache():
"""Initialize FastAPI Cache with in-memory backend"""
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))
raise
async def invalidate_cache_key(key: str):
"""Invalidate a specific cache key"""
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
Args:
expire (int): Cache TTL in seconds
key_builder (callable, optional): Custom key builder function
"""
return cache(
expire=expire,
key_builder=key_builder
) |