"""CDEK HTTP client and provider adapter.""" import asyncio from collections.abc import Awaitable, Callable from typing import Any import httpx from app.adapters.delivery_providers.base import DeliveryProvider from app.adapters.delivery_providers.cdek.auth import CDEKAuthClient from app.adapters.delivery_providers.cdek.mapper import map_cdek_response from app.config import AdapterConfig from app.schemas.request import DeliveryRequest from app.schemas.response import DeliveryPrice class CDEKClientError(RuntimeError): """Raised when CDEK tariff request fails.""" class CDEKClient: def __init__( self, http_client: httpx.AsyncClient, auth_client: CDEKAuthClient, *, base_url: str, timeout_seconds: float = 10.0, retry_attempts: int = 2, retry_backoff_seconds: float = 0.2, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, ) -> None: self._http_client = http_client self._auth_client = auth_client normalized_base_url = base_url.rstrip("/") self._city_lookup_url = f"{normalized_base_url}/location/cities" self._tariff_url = f"{base_url.rstrip('/')}/calculator/tarifflist" self._timeout_seconds = timeout_seconds self._retry_attempts = retry_attempts self._retry_backoff_seconds = retry_backoff_seconds self._sleep = sleep self._city_code_cache: dict[str, int] = {} async def get_raw_price(self, request: DeliveryRequest) -> dict[str, Any]: payload = await self._build_payload(request) for attempt in range(self._retry_attempts + 1): try: token = await self._auth_client.get_access_token() response = await self._http_client.post( self._tariff_url, json=payload, headers={"Authorization": f"Bearer {token}"}, timeout=self._timeout_seconds, ) except (httpx.TimeoutException, httpx.TransportError) as exc: if attempt < self._retry_attempts: await self._sleep(self._retry_delay(attempt)) continue raise CDEKClientError( "CDEK tariff request failed after retry attempts." ) from exc if self._should_retry(response.status_code): if attempt < self._retry_attempts: await self._sleep(self._retry_delay(attempt)) continue raise CDEKClientError( f"CDEK tariff request failed with status {response.status_code}." ) try: response.raise_for_status() raw_payload = response.json() except (httpx.HTTPError, TypeError, ValueError) as exc: raise CDEKClientError("CDEK tariff request returned invalid payload.") from exc if not isinstance(raw_payload, dict): raise CDEKClientError("CDEK tariff payload must be a JSON object.") return raw_payload raise CDEKClientError("CDEK tariff request failed unexpectedly.") def _retry_delay(self, attempt: int) -> float: return self._retry_backoff_seconds * (attempt + 1) @staticmethod def _should_retry(status_code: int) -> bool: return status_code == 429 or status_code >= 500 async def _build_payload(self, request: DeliveryRequest) -> dict[str, Any]: from_city_code = await self._resolve_city_code(request.from_city) to_city_code = await self._resolve_city_code(request.to_city) return { "from_location": {"code": from_city_code}, "to_location": {"code": to_city_code}, "packages": [ { "weight": int(round(request.weight_kg * 1000)), "length": int(round(request.length_cm)), "width": int(round(request.width_cm)), "height": int(round(request.height_cm)), } ], } async def _resolve_city_code(self, city: str) -> int: normalized_city = city.strip().casefold() cached_code = self._city_code_cache.get(normalized_city) if cached_code is not None: return cached_code for attempt in range(self._retry_attempts + 1): try: token = await self._auth_client.get_access_token() response = await self._http_client.get( self._city_lookup_url, params={"city": city, "country_codes": "RU", "size": 1}, headers={"Authorization": f"Bearer {token}"}, timeout=self._timeout_seconds, ) except (httpx.TimeoutException, httpx.TransportError) as exc: if attempt < self._retry_attempts: await self._sleep(self._retry_delay(attempt)) continue raise CDEKClientError( "CDEK city lookup request failed after retry attempts." ) from exc if self._should_retry(response.status_code): if attempt < self._retry_attempts: await self._sleep(self._retry_delay(attempt)) continue raise CDEKClientError( f"CDEK city lookup failed with status {response.status_code}." ) try: response.raise_for_status() body = response.json() except (httpx.HTTPError, TypeError, ValueError) as exc: raise CDEKClientError("CDEK city lookup returned invalid payload.") from exc break else: raise CDEKClientError("CDEK city lookup failed unexpectedly.") if not isinstance(body, list) or not body: raise CDEKClientError(f"CDEK city lookup returned no matches for '{city}'.") first_item = body[0] if not isinstance(first_item, dict): raise CDEKClientError("CDEK city lookup response entry must be an object.") raw_city_code = first_item.get("code") if raw_city_code is None: raise CDEKClientError("CDEK city lookup response has no city code.") try: city_code = int(raw_city_code) except (TypeError, ValueError) as exc: raise CDEKClientError("CDEK city lookup response city code is invalid.") from exc self._city_code_cache[normalized_city] = city_code return city_code class CDEKProvider(DeliveryProvider): name = "cdek" def __init__(self, client: CDEKClient, *, cache_ttl_seconds: int = 900) -> None: self._client = client self.cache_ttl_seconds = cache_ttl_seconds @classmethod def from_adapter_config( cls, *, http_client: httpx.AsyncClient, adapter_config: AdapterConfig, ) -> "CDEKProvider": auth_client = CDEKAuthClient( http_client=http_client, base_url=adapter_config.cdek_base_url, client_id=adapter_config.cdek_client_id, client_secret=adapter_config.cdek_client_secret, timeout_seconds=adapter_config.cdek_timeout_seconds, ) client = CDEKClient( http_client=http_client, auth_client=auth_client, base_url=adapter_config.cdek_base_url, timeout_seconds=adapter_config.cdek_timeout_seconds, retry_attempts=adapter_config.cdek_retry_attempts, retry_backoff_seconds=adapter_config.cdek_retry_backoff_seconds, ) return cls(client=client, cache_ttl_seconds=adapter_config.cdek_cache_ttl_seconds) async def get_price(self, request: DeliveryRequest) -> DeliveryPrice: raw_payload = await self._client.get_raw_price(request) return map_cdek_response(raw_payload)