From 082e5c11ec5698fe190167f32b6b818c2c56d35a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=A0=D0=B0=D0=B8=D1=81=20=D0=AE=D1=81=D1=83=D0=BF=D0=B0?= =?UTF-8?q?=D0=BB=D0=B8=D0=B5=D0=B2?= Date: Sat, 7 Mar 2026 15:38:26 +0300 Subject: [PATCH] 005 Implement service --- app/services/aggregator.py | 149 ++++++++++++++++- spec/index.md | 8 +- ...5_implement_aggregator_service_workflow.md | 2 +- tests/services/test_aggregator.py | 155 ++++++++++++++++++ 4 files changed, 306 insertions(+), 8 deletions(-) create mode 100644 tests/services/test_aggregator.py diff --git a/app/services/aggregator.py b/app/services/aggregator.py index 2a309f2..e3fd596 100644 --- a/app/services/aggregator.py +++ b/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 - raise NotImplementedError("Aggregator workflow is implemented in task 005.") + 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) + ] + 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) diff --git a/spec/index.md b/spec/index.md index 5161475..b3391a3 100644 --- a/spec/index.md +++ b/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** -- DONE: **5** +- TODO: **4** +- DONE: **6** diff --git a/spec/tasks/005_implement_aggregator_service_workflow.md b/spec/tasks/005_implement_aggregator_service_workflow.md index 5109e9b..ea254c8 100644 --- a/spec/tasks/005_implement_aggregator_service_workflow.md +++ b/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 --- diff --git a/tests/services/test_aggregator.py b/tests/services/test_aggregator.py new file mode 100644 index 0000000..3e7274c --- /dev/null +++ b/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"] +