005 Implement service

app/services/aggregator.py

@@ -1,10 +1,153 @@
-"""Aggregator service orchestrator skeleton."""
+"""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
+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 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]:
-        _ = request
+        normalized_request = normalize_delivery_request(
-        raise NotImplementedError("Aggregator workflow is implemented in task 005.")
+            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)
+        ]
+        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,
+            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)

spec/index.md

@@ -1,7 +1,7 @@
 # Spec Tasks Index
 
 > ⚠️ This file is generated. Do not edit manually.
-> Generated at (UTC): `2026-03-07T12:18:12+00:00`
+> Generated at (UTC): `2026-03-07T12:38:07+00:00`
 
 ## Tasks
 
@@ -12,7 +12,7 @@
 | 002 | DONE | 2026-03-07 | Implement pure domain price rules | `spec/tasks/002_implement_pure_domain_quote_rules.md` |
 | 003 | DONE | 2026-03-07 | Add CDEK provider adapter | `spec/tasks/003_add_cdek_provider_adapter.md` |
 | 004 | DONE | 2026-03-07 | Add Redis price cache repository | `spec/tasks/004_add_redis_price_cache_repository.md` |
-| 005 | TODO | 2026-03-07 | Implement aggregator service workflow | `spec/tasks/005_implement_aggregator_service_workflow.md` |
+| 005 | DONE | 2026-03-07 | Implement aggregator service workflow | `spec/tasks/005_implement_aggregator_service_workflow.md` |
 | 006 | TODO | 2026-03-07 | Add delivery price controller endpoint | `spec/tasks/006_add_delivery_price_controller.md` |
 | 007 | TODO | 2026-03-07 | Add observability correlation and telemetry | `spec/tasks/007_add_observability_correlation.md` |
 | 008 | TODO | 2026-03-07 | Add SigNoz to Telegram alerting configuration | `spec/tasks/008_add_signoz_telegram_alerting.md` |
@@ -21,5 +21,5 @@
 ## Summary
 
 - Total: **10**
-- TODO: **5**
+- TODO: **4**
-- DONE: **5**
+- DONE: **6**

spec/tasks/005_implement_aggregator_service_workflow.md

@@ -1,7 +1,7 @@
 ---
 id: 005
 title: Implement aggregator service workflow
-status: TODO
+status: DONE
 created: 2026-03-07
 ---
 

tests/services/test_aggregator.py

@@ -0,0 +1,155 @@
+import asyncio
+from decimal import Decimal
+
+from app.schemas.request import DeliveryEntity, DeliveryRequest
+from app.schemas.response import DeliveryPrice
+from app.services.aggregator import AggregatorService
+
+
+_MISSING = object()
+
+
+class StubProvider:
+    def __init__(
+        self,
+        name: str,
+        *,
+        response: DeliveryPrice | None = None,
+        error: Exception | None = None,
+        cache_ttl_seconds: int = 900,
+    ) -> None:
+        self.name = name
+        self._response = response
+        self._error = error
+        self.cache_ttl_seconds = cache_ttl_seconds
+        self.calls: list[DeliveryRequest] = []
+
+    async def get_price(self, request: DeliveryRequest) -> DeliveryPrice:
+        self.calls.append(request)
+        if self._error is not None:
+            raise self._error
+        if self._response is None:
+            raise RuntimeError("Stub provider has no response configured.")
+        return self._response
+
+
+class StubCache:
+    def __init__(
+        self,
+        *,
+        forced_get_value: object = _MISSING,
+    ) -> None:
+        self._storage: dict[str, object] = {}
+        self._forced_get_value = forced_get_value
+        self.get_calls: list[str] = []
+        self.set_calls: list[tuple[str, object, int | None]] = []
+
+    async def get(self, key: str) -> object | None:
+        self.get_calls.append(key)
+        if self._forced_get_value is not _MISSING:
+            return self._forced_get_value
+        return self._storage.get(key)
+
+    async def set(self, key: str, value: object, ttl: int | None = None) -> None:
+        self.set_calls.append((key, value, ttl))
+        self._storage[key] = value
+
+
+def _make_request() -> DeliveryRequest:
+    return DeliveryRequest(
+        entity=DeliveryEntity.INDIVIDUAL,
+        from_city="Moscow",
+        to_city="Kazan",
+        weight_kg=1.234,
+        length_cm=30.0,
+        width_cm=20.0,
+        height_cm=10.0,
+    )
+
+
+def _make_price(provider: str, price: str) -> DeliveryPrice:
+    return DeliveryPrice(
+        provider=provider,
+        service_name="standard",
+        price=Decimal(price),
+        currency="RUB",
+        delivery_days_min=2,
+        delivery_days_max=4,
+    )
+
+
+def test_get_all_prices_full_success_returns_sorted_and_updates_cache() -> None:
+    provider_a = StubProvider(name="a", response=_make_price("a", "300.00"), cache_ttl_seconds=111)
+    provider_b = StubProvider(name="b", response=_make_price("b", "100.00"), cache_ttl_seconds=222)
+    cache = StubCache()
+    service = AggregatorService([provider_a, provider_b], cache=cache)
+
+    result = asyncio.run(service.get_all_prices(_make_request()))
+
+    assert [price.provider for price in result] == ["b", "a"]
+    assert len(provider_a.calls) == 1
+    assert len(provider_b.calls) == 1
+    assert len(cache.get_calls) == 2
+    assert [ttl for _, _, ttl in cache.set_calls] == [111, 222]
+
+
+def test_get_all_prices_partial_failure_excludes_failed_provider() -> None:
+    provider_ok = StubProvider(name="ok", response=_make_price("ok", "150.00"))
+    provider_failed = StubProvider(name="failed", error=RuntimeError("provider unavailable"))
+    service = AggregatorService([provider_ok, provider_failed], cache=StubCache())
+
+    result = asyncio.run(service.get_all_prices(_make_request()))
+
+    assert [price.provider for price in result] == ["ok"]
+    assert len(provider_ok.calls) == 1
+    assert len(provider_failed.calls) == 1
+
+
+def test_get_all_prices_all_failures_returns_empty_list() -> None:
+    provider_a = StubProvider(name="a", error=RuntimeError("a down"))
+    provider_b = StubProvider(name="b", error=RuntimeError("b down"))
+    service = AggregatorService([provider_a, provider_b], cache=StubCache())
+
+    result = asyncio.run(service.get_all_prices(_make_request()))
+
+    assert result == []
+    assert len(provider_a.calls) == 1
+    assert len(provider_b.calls) == 1
+
+
+def test_get_all_prices_cache_hit_skips_provider_call() -> None:
+    cached_payload = _make_price("cdek", "99.00").model_dump(mode="json")
+    cache = StubCache(forced_get_value=cached_payload)
+    provider = StubProvider(name="cdek", response=_make_price("cdek", "150.00"))
+    service = AggregatorService([provider], cache=cache)
+
+    result = asyncio.run(service.get_all_prices(_make_request()))
+
+    assert [price.provider for price in result] == ["cdek"]
+    assert result[0].price == Decimal("99.00")
+    assert provider.calls == []
+    assert len(cache.get_calls) == 1
+    assert cache.set_calls == []
+
+
+def test_get_all_prices_delegates_filtering_and_sorting_to_domain_logic() -> None:
+    provider_a = StubProvider(name="a", response=_make_price("a", "300.00"))
+    provider_b = StubProvider(name="b", response=_make_price("b", "100.00"))
+    delegated_inputs: list[list[DeliveryPrice]] = []
+
+    def fake_filter_and_sort(prices):
+        price_list = list(prices)
+        delegated_inputs.append(price_list)
+        return [price_list[0]]
+
+    service = AggregatorService(
+        [provider_a, provider_b],
+        cache=StubCache(),
+        filter_and_sort_prices_fn=fake_filter_and_sort,
+    )
+
+    result = asyncio.run(service.get_all_prices(_make_request()))
+
+    assert [price.provider for price in delegated_inputs[0]] == ["a", "b"]
+    assert [price.provider for price in result] == ["a"]
+