Раис Юсупалиев
291 lines
12 KiB
Python
291 lines
12 KiB
Python
"""CDEK HTTP client and provider adapter."""
|
|
|
|
import asyncio
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
import logging
|
|
|
|
import httpx
|
|
|
|
from app.adapters.delivery_providers.base import DeliveryProvider, ProviderRequestError
|
|
from app.adapters.delivery_providers.cdek.auth import CDEKAuthClient
|
|
from app.adapters.delivery_providers.cdek.mapper import map_cdek_response
|
|
from app.adapters.delivery_providers.cdek.order_mapper import (
|
|
CDEKOrderMappingError,
|
|
map_cdek_order_request,
|
|
map_cdek_order_response,
|
|
)
|
|
from app.config import AdapterConfig
|
|
from app.schemas.order import OrderCreateRequest, OrderCreateResponse
|
|
from app.schemas.request import DeliveryRequest
|
|
from app.schemas.response import DeliveryPrice
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class CDEKClientError(RuntimeError):
|
|
"""Raised when CDEK tariff request fails."""
|
|
|
|
|
|
class CDEKRequestError(CDEKClientError, ProviderRequestError):
|
|
"""Raised when CDEK rejects request data as invalid."""
|
|
|
|
|
|
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/suggest/cities"
|
|
self._tariff_url = f"{normalized_base_url}/calculator/tarifflist"
|
|
self._orders_url = f"{normalized_base_url}/orders"
|
|
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.")
|
|
|
|
async def register_order(self, request: OrderCreateRequest) -> OrderCreateResponse:
|
|
payload = map_cdek_order_request(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._orders_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 order registration 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(
|
|
"CDEK order registration failed with retriable status "
|
|
f"{response.status_code}."
|
|
)
|
|
|
|
if 400 <= response.status_code < 500:
|
|
raise CDEKRequestError(
|
|
"CDEK order registration request was rejected with status "
|
|
f"{response.status_code}."
|
|
)
|
|
|
|
try:
|
|
response.raise_for_status()
|
|
raw_payload = response.json()
|
|
except (httpx.HTTPError, TypeError, ValueError) as exc:
|
|
raise CDEKClientError(
|
|
"CDEK order registration returned invalid payload."
|
|
) from exc
|
|
|
|
if not isinstance(raw_payload, dict):
|
|
raise CDEKClientError(
|
|
"CDEK order registration payload must be a JSON object."
|
|
)
|
|
|
|
try:
|
|
return map_cdek_order_response(raw_payload)
|
|
except CDEKOrderMappingError as exc:
|
|
raise CDEKClientError(
|
|
"CDEK order registration response payload is invalid."
|
|
) from exc
|
|
|
|
raise CDEKClientError("CDEK order registration 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,
|
|
country_code=request.country_code,
|
|
)
|
|
to_city_code = await self._resolve_city_code(
|
|
request.to_city,
|
|
country_code=request.country_code,
|
|
)
|
|
return {
|
|
"type": 2,
|
|
"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, *, country_code: str | None) -> int:
|
|
normalized_city = city.strip().casefold()
|
|
normalized_country_code = (
|
|
country_code.strip().upper() if country_code is not None else ""
|
|
)
|
|
cache_key = f"{normalized_country_code}:{normalized_city}"
|
|
cached_code = self._city_code_cache.get(cache_key)
|
|
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()
|
|
params = {"name": city}
|
|
if normalized_country_code:
|
|
params["country_code"] = normalized_country_code
|
|
response = await self._http_client.get(
|
|
self._city_lookup_url,
|
|
params=params,
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
timeout=self._timeout_seconds,
|
|
)
|
|
log.info(response.json())
|
|
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 isinstance(body, dict):
|
|
first_item = body
|
|
elif isinstance(body, list) and body:
|
|
first_item = body[0]
|
|
elif isinstance(body, list):
|
|
raise CDEKRequestError(
|
|
f"CDEK city lookup returned no matches for '{city}'."
|
|
)
|
|
else:
|
|
raise CDEKClientError("CDEK city lookup response must be an object or array.")
|
|
|
|
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[cache_key] = 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)
|
|
|
|
async def register_order(self, request: OrderCreateRequest) -> OrderCreateResponse:
|
|
return await self._client.register_order(request)
|