"""Aggregator service orchestrating providers, cache and domain rules.""" import asyncio import hashlib import json from collections.abc import Callable, Iterable, Sequence from typing import Protocol from app.adapters.delivery_providers.base import DeliveryProvider, ProviderRequestError from app.domain.price import ( DEFAULT_WEIGHT_ROUND_SCALE, NormalizedDeliveryRequest, filter_and_sort_prices, normalize_delivery_request, ) from app.schemas.request import DeliveryRequest from app.schemas.response import DeliveryPrice class AggregatorServiceError(RuntimeError): """Base exception for AggregatorService failures.""" class InvalidDeliveryRequestError(AggregatorServiceError): """Raised when provider rejects delivery request as invalid.""" class PriceCacheProtocol(Protocol): async def get(self, key: str) -> object | None: ... async def set(self, key: str, value: object, ttl: int | None = None) -> None: ... class AggregatorService: def __init__( self, providers: Sequence[DeliveryProvider], cache: PriceCacheProtocol | None = None, *, weight_round_scale: int = DEFAULT_WEIGHT_ROUND_SCALE, filter_and_sort_prices_fn: Callable[[Iterable[object]], list[object]] = ( filter_and_sort_prices ), ) -> None: self._providers = tuple(providers) self._cache = cache self._weight_round_scale = weight_round_scale self._filter_and_sort_prices = filter_and_sort_prices_fn async def get_all_prices(self, request: DeliveryRequest) -> list[DeliveryPrice]: normalized_request = normalize_delivery_request( request, weight_round_scale=self._weight_round_scale ) provider_request = self._to_provider_request(normalized_request) provider_results = await asyncio.gather( *( self._get_provider_price( provider=provider, request=provider_request, cache_key=self._build_cache_key( provider_name=provider.name, request=normalized_request, ), ) for provider in self._providers ), return_exceptions=True, ) successful_results = [ price for price in provider_results if isinstance(price, DeliveryPrice) ] provider_errors = [ error for error in provider_results if isinstance(error, Exception) ] if ( not successful_results and provider_errors and any(isinstance(error, ProviderRequestError) for error in provider_errors) ): raise InvalidDeliveryRequestError( "Delivery request is invalid for configured providers." ) filtered_and_sorted = self._filter_and_sort_prices(successful_results) return [self._coerce_delivery_price(price) for price in filtered_and_sorted] async def _get_provider_price( self, *, provider: DeliveryProvider, request: DeliveryRequest, cache_key: str, ) -> DeliveryPrice: cached_price = await self._get_cached_price(cache_key) if cached_price is not None: return cached_price fresh_price = await provider.get_price(request) await self._set_cached_price( cache_key, fresh_price, ttl=getattr(provider, "cache_ttl_seconds", None), ) return fresh_price async def _get_cached_price(self, cache_key: str) -> DeliveryPrice | None: if self._cache is None: return None try: payload = await self._cache.get(cache_key) except Exception: return None if payload is None: return None try: return self._coerce_delivery_price(payload) except Exception: return None async def _set_cached_price( self, cache_key: str, payload: DeliveryPrice, *, ttl: int | None, ) -> None: if self._cache is None: return try: await self._cache.set(cache_key, payload, ttl=ttl) except Exception: return @staticmethod def _to_provider_request(request: NormalizedDeliveryRequest) -> DeliveryRequest: return DeliveryRequest( entity=request.entity, from_city=request.from_city, to_city=request.to_city, country_code=request.country_code, weight_kg=request.weight_kg, length_cm=request.length_cm, width_cm=request.width_cm, height_cm=request.height_cm, ) @staticmethod def _build_cache_key(provider_name: str, request: NormalizedDeliveryRequest) -> str: cache_payload = { "provider": provider_name, "entity": request.entity, "from_city": request.from_city, "to_city": request.to_city, "country_code": request.country_code, "weight_kg": str(request.weight_kg), "length_cm": str(request.length_cm), "width_cm": str(request.width_cm), "height_cm": str(request.height_cm), } serialized_payload = json.dumps( cache_payload, ensure_ascii=True, sort_keys=True, separators=(",", ":"), ) cache_hash = hashlib.sha256(serialized_payload.encode("utf-8")).hexdigest() return f"delivery-price:{provider_name}:{cache_hash}" @staticmethod def _coerce_delivery_price(value: object) -> DeliveryPrice: return DeliveryPrice.model_validate(value, from_attributes=True)