g2s-aggregator/app/repositories/cache/redis_cache.py

"""Redis-backed cache repository."""

from dataclasses import asdict, is_dataclass
from decimal import Decimal
from enum import Enum
import json
from typing import Any, Callable, Protocol

from pydantic import BaseModel
import structlog

from app.config import RepositoryConfig


logger = structlog.get_logger(__name__)


class RedisClientProtocol(Protocol):
    async def get(self, key: str) -> bytes | str | None: ...

    async def set(self, key: str, value: str, ex: int | None = None) -> Any: ...

    async def delete(self, key: str) -> Any: ...


class CacheMetricsProtocol(Protocol):
    def record_request(self, *, operation: str, outcome: str) -> None: ...

    def record_error(self, *, operation: str, error_type: str) -> None: ...

    def record_hit(self, *, hit: bool) -> None: ...


class PriceCacheRepositoryError(RuntimeError):
    """Deterministic repository error raised on Redis operation failures."""


class PriceCache:
    """Repository for cached provider payloads."""

    def __init__(
        self,
        redis_client: RedisClientProtocol,
        *,
        ttl_seconds: int,
        metrics: CacheMetricsProtocol | None = None,
    ) -> None:
        self._redis_client = redis_client
        self._ttl_seconds = ttl_seconds
        self._metrics = metrics

    @classmethod
    def from_repository_config(
        cls,
        repository_config: RepositoryConfig,
        *,
        client_factory: Callable[[str], RedisClientProtocol] | None = None,
        metrics: CacheMetricsProtocol | None = None,
    ) -> "PriceCache":
        resolved_factory = client_factory or _default_redis_client_factory
        redis_client = resolved_factory(repository_config.redis_dsn)
        return cls(
            redis_client=redis_client,
            ttl_seconds=repository_config.price_cache_ttl_seconds,
            metrics=metrics,
        )

    async def get(self, key: str) -> Any | None:
        try:
            payload = await self._redis_client.get(key)
        except Exception as exc:
            self._record_failure(operation="get", error=exc)
            raise PriceCacheRepositoryError("Redis cache read failed.") from exc
        if payload is None:
            self._record_request(operation="get", outcome="miss")
            self._record_hit(hit=False)
            return None

        if isinstance(payload, bytes):
            serialized_payload = payload.decode("utf-8")
        elif isinstance(payload, str):
            serialized_payload = payload
        else:
            raise TypeError("Redis cache payload must be bytes or str.")

        result = json.loads(serialized_payload)
        self._record_request(operation="get", outcome="hit")
        self._record_hit(hit=True)
        return result

    async def set(self, key: str, value: Any, ttl: int | None = None) -> None:
        serialized_payload = _serialize(value)
        resolved_ttl = self._ttl_seconds if ttl is None else ttl
        try:
            await self._redis_client.set(key, serialized_payload, ex=resolved_ttl)
        except Exception as exc:
            self._record_failure(operation="set", error=exc)
            raise PriceCacheRepositoryError("Redis cache write failed.") from exc
        self._record_request(operation="set", outcome="success")

    async def invalidate(self, key: str) -> None:
        try:
            await self._redis_client.delete(key)
        except Exception as exc:
            self._record_failure(operation="invalidate", error=exc)
            raise PriceCacheRepositoryError("Redis cache invalidate failed.") from exc
        self._record_request(operation="invalidate", outcome="success")

    def _record_failure(self, *, operation: str, error: Exception) -> None:
        error_type = type(error).__name__
        logger.warning(
            "cache.operation_failed",
            operation=operation,
            error_type=error_type,
        )
        if self._metrics is not None:
            self._metrics.record_request(operation=operation, outcome="error")
            self._metrics.record_error(operation=operation, error_type=error_type)

    def _record_request(self, *, operation: str, outcome: str) -> None:
        if self._metrics is not None:
            self._metrics.record_request(operation=operation, outcome=outcome)

    def _record_hit(self, *, hit: bool) -> None:
        if self._metrics is not None:
            self._metrics.record_hit(hit=hit)


def _default_redis_client_factory(redis_dsn: str) -> RedisClientProtocol:
    """Create async Redis client from DSN."""

    try:
        from aioredis import from_url as aioredis_from_url  # type: ignore
    except ModuleNotFoundError:
        try:
            import redis.asyncio as redis_asyncio  # type: ignore[import-not-found]
        except ModuleNotFoundError as exc:
            raise RuntimeError(
                "No async Redis client implementation is available."
            ) from exc
        return redis_asyncio.from_url(redis_dsn, encoding="utf-8", decode_responses=False)

    return aioredis_from_url(redis_dsn, encoding="utf-8", decode_responses=False)


def _serialize(value: Any) -> str:
    normalized = _normalize_for_json(value)
    return json.dumps(normalized, ensure_ascii=True, sort_keys=True, separators=(",", ":"))


def _normalize_for_json(value: Any) -> Any:
    if isinstance(value, BaseModel):
        return _normalize_for_json(value.model_dump(mode="json"))
    if is_dataclass(value):
        return _normalize_for_json(asdict(value))
    if isinstance(value, Enum):
        return _normalize_for_json(value.value)
    if isinstance(value, Decimal):
        return str(value)
    if isinstance(value, dict):
        return {str(key): _normalize_for_json(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [_normalize_for_json(item) for item in value]
    if value is None or isinstance(value, (str, int, float, bool)):
        return value
    raise TypeError(f"Unsupported cache payload type: {type(value)!r}")