g2s-aggregator/app/services/aggregator.py

"""Aggregator service orchestrating providers, cache and domain rules."""

import asyncio
import hashlib
import json
from collections.abc import Iterable, Sequence
from decimal import Decimal
from typing import Protocol

from app.adapters.delivery_providers.base import DeliveryProvider, ProviderRequestError
from app.domain.price import (
    DEFAULT_PROVIDER_PRICE_MULTIPLIER,
    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 FilterAndSortPricesFn(Protocol):
    def __call__(
        self,
        prices: Iterable[object],
        *,
        price_multiplier: Decimal = DEFAULT_PROVIDER_PRICE_MULTIPLIER,
    ) -> list[object]: ...


class AggregatorService:
    def __init__(
        self,
        providers: Sequence[DeliveryProvider],
        cache: PriceCacheProtocol | 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._weight_round_scale = weight_round_scale
        self._provider_price_multiplier = provider_price_multiplier
        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,
            price_multiplier=self._provider_price_multiplier,
        )
        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)