"""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}")