docs/standards/src/templates/app/infra/http_cache.py

"""# 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 functools import wraps
from typing import Any, Callable, Optional

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) -> Optional[str]:
        """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: Optional[int] = 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: Optional[HttpCacheStore] = 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