File size: 2,573 Bytes
e7abd9e 23c96f8 5c2e213 e7abd9e 58582d3 5c2e213 58582d3 5c2e213 58582d3 e7abd9e 5c2e213 58582d3 5c2e213 58582d3 5c2e213 e7abd9e 58582d3 5c2e213 e7abd9e 58582d3 e7abd9e 58582d3 5c2e213 e7abd9e 58582d3 e7abd9e 58582d3 e7abd9e 58582d3 |
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 77 78 79 |
from fastapi_cache import FastAPICache
from fastapi_cache.backends.inmemory import InMemoryBackend
from fastapi_cache.decorator import cache
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)
|