"""Aggregator service orchestrating providers, cache and domain rules.""" import asyncio import hashlib import json from collections.abc import Iterable, Mapping, Sequence from contextlib import AbstractAsyncContextManager from decimal import Decimal from typing import Protocol import structlog from app.adapters.address_suggestions.base import ( AddressSuggestionClientError, AddressSuggestionProvider, AddressSuggestionRequestError, ) from app.adapters.delivery_providers.base import DeliveryProvider, ProviderRequestError from app.adapters.tbank.base import ( TBankPaymentAdapterError, TBankPaymentRequestError, ) from app.domain.price import ( DEFAULT_PROVIDER_PRICE_MULTIPLIER, DEFAULT_WEIGHT_ROUND_SCALE, NormalizedDeliveryRequest, filter_and_sort_prices, normalize_delivery_request, ) from app.repositories.order import OrderData from app.schemas.payment import InitPaymentRequest, InitPaymentResponse from app.schemas.request import AddressSuggestRequest, DeliveryCalculationRequest from app.schemas.response import AddressSuggestion, DeliveryPrice logger = structlog.get_logger(__name__) class AggregatorServiceError(RuntimeError): """Base exception for AggregatorService failures.""" class InvalidDeliveryRequestError(AggregatorServiceError): """Raised when provider rejects delivery request as invalid.""" class UnsupportedAddressSuggestionCountryError(AggregatorServiceError): """Raised when address suggestions are not configured for a country.""" class InvalidAddressSuggestRequestError(AggregatorServiceError): """Raised when provider rejects address suggestion payload.""" class AddressSuggestionsUnavailableError(AggregatorServiceError): """Raised when address suggestions cannot be completed.""" class InvalidInitPaymentRequestError(AggregatorServiceError): """Raised when provider rejects payment init payload as invalid.""" class InitPaymentUnavailableError(AggregatorServiceError): """Raised when payment initialization cannot be completed.""" 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 PaymentAdapterProtocol(Protocol): async def create_payment_link(self, order_uuid: str, amount_kopecks: int) -> str: ... class OrderRepositoryProtocol(Protocol): def session(self) -> AbstractAsyncContextManager[object]: ... async def create_order(self, session: object, order_data: OrderData) -> object: ... class FilterAndSortPricesFn(Protocol): def __call__( self, prices: Iterable[object], *, price_multiplier: Decimal = DEFAULT_PROVIDER_PRICE_MULTIPLIER, parcel_type: object | None = None, ) -> list[object]: ... class AggregatorService: def __init__( self, providers: Sequence[DeliveryProvider], cache: PriceCacheProtocol | None = None, payment_adapter: PaymentAdapterProtocol | None = None, order_repository: OrderRepositoryProtocol | None = None, address_suggestion_providers: Sequence[AddressSuggestionProvider] = (), address_suggestion_country_to_provider: Mapping[str, str] | None = None, *, weight_round_scale: int = DEFAULT_WEIGHT_ROUND_SCALE, provider_price_multiplier: Decimal = DEFAULT_PROVIDER_PRICE_MULTIPLIER, filter_and_sort_prices_fn: FilterAndSortPricesFn = filter_and_sort_prices, ) -> None: self._providers = tuple(providers) self._cache = cache self._payment_adapter = payment_adapter self._order_repository = order_repository self._weight_round_scale = weight_round_scale self._provider_price_multiplier = provider_price_multiplier self._filter_and_sort_prices = filter_and_sort_prices_fn self._address_suggestion_providers: dict[str, AddressSuggestionProvider] = { provider.name: provider for provider in address_suggestion_providers } self._address_suggestion_country_to_provider = ( address_suggestion_country_to_provider or {} ) async def get_all_prices( self, request: DeliveryCalculationRequest ) -> 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_prices( 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 provider_prices in provider_results if isinstance(provider_prices, list) for price in provider_prices ] 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, price_multiplier=self._provider_price_multiplier, parcel_type=request.parcel_type, ) return [self._coerce_delivery_price(price) for price in filtered_and_sorted] async def suggest_addresses( self, request: AddressSuggestRequest ) -> list[AddressSuggestion]: provider = self._resolve_address_suggestion_provider(request.country_code) try: suggestions = await provider.suggest(request) except AddressSuggestionRequestError as exc: raise InvalidAddressSuggestRequestError( "Address suggestion request is invalid for the configured provider." ) from exc except AddressSuggestionClientError as exc: raise AddressSuggestionsUnavailableError( "Address suggestions are temporarily unavailable." ) from exc return [self._coerce_address_suggestion(item) for item in suggestions] async def init_payment(self, request: InitPaymentRequest) -> InitPaymentResponse: if self._payment_adapter is None: raise InitPaymentUnavailableError("Payment adapter is not configured.") try: payment_url = await self._payment_adapter.create_payment_link( order_uuid=request.order_uuid, amount_kopecks=request.price, ) except TBankPaymentRequestError as exc: logger.exception( "payment_init_rejected", order_uuid=request.order_uuid, provider_status_code=exc.status_code, provider_error_code=exc.error_code, provider_error_message=exc.provider_message, provider_error_details=exc.details, ) raise InvalidInitPaymentRequestError( "Payment init request is invalid for the configured provider." ) from exc except TBankPaymentAdapterError as exc: raise InitPaymentUnavailableError( "Payment initialization is temporarily unavailable." ) from exc except Exception as exc: raise InitPaymentUnavailableError( "Payment initialization is temporarily unavailable." ) from exc await self._persist_order(request=request, payment_url=payment_url) return InitPaymentResponse(payment_url=payment_url) async def _persist_order( self, *, request: InitPaymentRequest, payment_url: str, ) -> None: if self._order_repository is None: return try: async with self._order_repository.session() as session: await self._order_repository.create_order( session, self._to_order_data(request=request, payment_url=payment_url), ) except Exception: logger.exception( "order_persistence_failed", order_uuid=request.order_uuid, ) @staticmethod def _to_order_data( *, request: InitPaymentRequest, payment_url: str, ) -> OrderData: payload = request.model_dump(mode="json") return OrderData( order_uuid=request.order_uuid, payment_url=payment_url, price=request.price, delivery_type=request.type, tariff_code=request.tariff_code, sender=payload["sender"], recipient=payload["recipient"], from_location=payload["from_location"], to_location=payload["to_location"], packages=payload["packages"], services=payload["services"], comment=request.comment, ) async def _get_provider_prices( self, *, provider: DeliveryProvider, request: DeliveryCalculationRequest, cache_key: str, ) -> list[DeliveryPrice]: cached_prices = await self._get_cached_prices(cache_key) if cached_prices is not None: return cached_prices fresh_prices = await provider.get_prices(request) await self._set_cached_prices( cache_key, fresh_prices, ttl=getattr(provider, "cache_ttl_seconds", None), ) return fresh_prices async def _get_cached_prices(self, cache_key: str) -> list[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_prices(payload) except Exception: return None async def _set_cached_prices( self, cache_key: str, payload: list[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, ) -> DeliveryCalculationRequest: return DeliveryCalculationRequest( entity=request.entity, from_city=request.from_city, to_city=request.to_city, 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, "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) @classmethod def _coerce_delivery_prices(cls, value: object) -> list[DeliveryPrice]: if not isinstance(value, list): raise TypeError("Cached delivery prices must be a list.") return [cls._coerce_delivery_price(item) for item in value] @staticmethod def _coerce_address_suggestion(value: object) -> AddressSuggestion: return AddressSuggestion.model_validate(value, from_attributes=True) def _resolve_address_suggestion_provider( self, country_code: str ) -> AddressSuggestionProvider: provider_id = self._address_suggestion_country_to_provider.get(country_code) if provider_id is None: raise UnsupportedAddressSuggestionCountryError( "Address suggestions are not configured for the requested country." ) provider = self._address_suggestion_providers.get(provider_id) if provider is None: raise UnsupportedAddressSuggestionCountryError( "Address suggestions are not configured for the requested country." ) return provider