Spaces:
Running
Running
File size: 3,458 Bytes
ee472e7 5b89071 ee472e7 5b89071 ee472e7 5b89071 ee472e7 5b89071 ee472e7 5b89071 ee472e7 5b89071 ee472e7 5b89071 ee472e7 5b89071 ee472e7 5b89071 ee472e7 5b89071 | 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 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | from datetime import timedelta
from app.config import CACHE_TTL
import logging
from app.core.formatting import LogFormatter
from typing import Optional, Any
from functools import wraps
logger = logging.getLogger(__name__)
# Try to import fastapi-cache2; fall back to no-op if unavailable
try:
from fastapi_cache import FastAPICache
from fastapi_cache.backends.inmemory import InMemoryBackend
from fastapi_cache.decorator import cache as _cache_decorator
CACHE_AVAILABLE = True
except ImportError:
logger.warning("fastapi-cache2 not available; caching disabled")
CACHE_AVAILABLE = False
if CACHE_AVAILABLE:
class CustomInMemoryBackend(InMemoryBackend):
def __init__(self):
"""Initialize the cache backend"""
super().__init__()
self._store = {}
async def delete(self, key: str) -> bool:
"""Delete a key from the cache"""
try:
if key in self._store:
del self._store[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._store.get(key)
async def set(self, key: str, value: Any, expire: Optional[int] = None) -> None:
"""Set a value in the cache"""
self._store[key] = value
def setup_cache():
"""Initialize FastAPI Cache with in-memory backend"""
if not CACHE_AVAILABLE:
logger.warning(LogFormatter.warning("Cache setup skipped — fastapi-cache2 not installed"))
return
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))
# Don't raise — let the app run without cache
async def invalidate_cache_key(key: str):
"""Invalidate a specific cache key"""
if not CACHE_AVAILABLE:
return
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.
Falls back to a no-op decorator when fastapi-cache2 is not installed.
"""
if CACHE_AVAILABLE:
return _cache_decorator(
expire=expire,
key_builder=key_builder
)
else:
# No-op decorator
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
return wrapper
return decorator
|