"""# Summary Domain-agnostic HTTP response cache. --- A thin Redis-backed response cache that wraps route handlers as a decorator. Example: ```python from infra.http_cache import cache_response @cache_response(ttl=60) async def list_items(request, skip: int = 0, limit: int = 100): ... ``` Notes: The cache operates directly on the endpoint and is independent of any repository-level cache. Cached responses expire via TTL. When no ``http_cache_store`` is wired on ``request.app.state.db_resources`` the decorator is a no-op pass-through, so applying/removing it is safe. """ import hashlib import json from collections.abc import Callable from functools import wraps from typing import Any from fastapi import Request, Response from fastapi.encoders import jsonable_encoder CACHE_PREFIX = "apicache:" NS_PREFIX = CACHE_PREFIX + "ns:" DEFAULT_TTL = 60 class HttpCacheStore: """Store string responses in Redis with TTL under the ``apicache:`` key space. Attributes: _redis (Redis): Asynchronous Redis client bound to the configured URL. """ def __init__(self, url: str = "redis://localhost:6379/0", decode_responses: bool = True): """Initialize the store with a Redis connection. Args: url (str): Redis connection URL. decode_responses (bool): Whether Redis returns decoded strings. Raises: ImportError: When the optional ``redis`` package is not installed. """ from redis.asyncio import Redis self._redis = Redis.from_url(url, decode_responses=decode_responses) async def get(self, key: str) -> str | None: """Read the value stored at ``key``. Args: key (str): Cache key to read. Returns: Optional[str]: Cached value, or ``None`` when absent. """ return await self._redis.get(CACHE_PREFIX + key) async def set(self, key: str, value: str, ttl: int) -> None: """Store ``value`` at ``key`` with a TTL. Args: key (str): Cache key to write. value (str): Serialized value to persist. ttl (int): Lifetime in seconds. """ await self._redis.set(CACHE_PREFIX + key, value, ex=ttl) async def delete(self, key: str) -> None: """Delete the value stored at ``key``. Args: key (str): Cache key to remove. """ await self._redis.delete(CACHE_PREFIX + key) async def generation(self, resource: str) -> str: """Return the current generation counter for a resource namespace. Args: resource (str): Resource namespace to read. Returns: str: Generation counter, or ``"0"`` when unset. """ value = await self._redis.get(NS_PREFIX + resource) return value if value is not None else "0" async def invalidate(self, *resources: str) -> None: """Invalidate all cached responses for the given resource namespaces. Args: *resources (str): Resource namespaces to invalidate. Notes: Bumps each namespace's generation counter so existing entries become unreachable immediately; Redis reclaims them at TTL expiry. """ if not resources: return pipe = self._redis.pipeline(transaction=False) for resource in resources: pipe.incr(NS_PREFIX + resource) await pipe.execute() async def close(self) -> None: """Close the underlying Redis connection.""" await self._redis.aclose() def _user_scope(request: Request) -> str: """Build a stable per-user cache scope. Args: request (Request): Incoming HTTP request. Returns: str: Scoped identifier, or ``""`` for auth-free endpoints. """ user = getattr(request.state, "user", None) if isinstance(user, dict): for field in ("username", "id", "email"): if user.get(field): return str(user[field]) token = getattr(request.state, "token", None) if token: return hashlib.sha256(str(token).encode()).hexdigest()[:16] return "" def _resource_name(request: Request) -> str: """Return the first path segment identifying the resource. Args: request (Request): Incoming HTTP request. Returns: str: First path segment (for example ``"items"``), or ``"root"`` when the path is empty. """ return request.url.path.strip("/").split("/", 1)[0] or "root" def _cache_key(request: Request, generation: str = "0") -> str: """Compute a stable cache key for a request. Args: request (Request): Incoming HTTP request. generation (str): Resource generation counter guarding stale entries. Returns: str: SHA-256 digest of the request method, path, query and path parameters, user scope, and generation. """ parts = [ request.method, request.url.path, "&".join(f"{k}={v}" for k, v in sorted(request.query_params.items())), "&".join(f"{k}={v}" for k, v in sorted(request.path_params.items())), _user_scope(request), generation, ] return hashlib.sha256("|".join(parts).encode()).hexdigest() def cache_response(ttl: int | None = None) -> Callable: """Decorate a GET-style route handler to cache its JSON response. Args: ttl (Optional[int]): Cache lifetime in seconds; defaults to ``DEFAULT_TTL``. Returns: Callable: Decorator that wraps the route handler. Notes: The wrapped handler must accept a ``request: Request`` parameter (as every openapi_first handler does). Responses that are already a ``fastapi.Response`` (for example 204) are returned untouched and never cached. When no ``http_cache_store`` is wired on ``request.app.state.db_resources`` the decorator passes through to the handler without caching. """ ttl = ttl if ttl is not None else DEFAULT_TTL def decorator(func: Callable) -> Callable: @wraps(func) async def wrapper(*args: Any, **kwargs: Any): request = kwargs.get("request") if request is None: request = next((a for a in args if isinstance(a, Request)), None) if request is None: return await func(*args, **kwargs) db_resources = getattr(request.app.state, "db_resources", None) or {} store: HttpCacheStore | None = db_resources.get("http_cache_store") if store is None: return await func(*args, **kwargs) key = _cache_key(request, await store.generation(_resource_name(request))) cached = await store.get(key) if cached is not None: return Response(content=cached, media_type="application/json") result = await func(*args, **kwargs) if isinstance(result, Response): return result payload = json.dumps( jsonable_encoder(result), separators=(",", ":"), ensure_ascii=False, ) await store.set(key, payload, ttl) return Response(content=payload, media_type="application/json") return wrapper return decorator