File size: 2,097 Bytes
1d31670 |
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 |
import asyncio
import json
import hashlib
from typing import Any, Callable, Dict, Optional
from functools import wraps
import logging
logger = logging.getLogger(__name__)
# Simple in-memory cache
_cache: Dict[str, Any] = {}
_cache_lock = asyncio.Lock()
def build_cache_key(namespace: str, *args, **kwargs) -> str:
"""Build a cache key from namespace and parameters"""
key_data = f"{namespace}:{args}:{sorted(kwargs.items())}"
return hashlib.md5(key_data.encode()).hexdigest()
def cached(expire: int = 300, key_builder: Optional[Callable] = None):
"""
Cache decorator for FastAPI endpoints
Args:
expire: Cache expiration time in seconds
key_builder: Function to build cache key
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
async def wrapper(*args, **kwargs):
# Build cache key
if key_builder:
cache_key = key_builder(func, **kwargs)
else:
cache_key = build_cache_key(func.__name__, *args, **kwargs)
# Check cache
async with _cache_lock:
if cache_key in _cache:
cached_data, timestamp = _cache[cache_key]
import time
if time.time() - timestamp < expire:
logger.debug(f"Cache hit for key: {cache_key}")
return cached_data
else:
# Expired, remove from cache
del _cache[cache_key]
# Cache miss, execute function
logger.debug(f"Cache miss for key: {cache_key}")
result = await func(*args, **kwargs)
# Store in cache
async with _cache_lock:
import time
_cache[cache_key] = (result, time.time())
return result
return wrapper
return decorator
def setup_cache():
"""Setup cache configuration"""
logger.info("FastAPI cache initialized with in-memory backend") |