Раис Юсупалиев
548 lines
20 KiB
Python
548 lines
20 KiB
Python
"""CDEK HTTP client and provider adapter."""
|
|
|
|
import asyncio
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import structlog
|
|
|
|
from app.adapters.delivery_providers.base import (
|
|
DeliveryProvider,
|
|
ProviderClientError,
|
|
ProviderRequestError,
|
|
)
|
|
from app.adapters.delivery_providers.cdek.auth import CDEKAuthClient
|
|
from app.adapters.delivery_providers.cdek.mapper import (
|
|
CDEKMappingError,
|
|
map_cdek_response,
|
|
map_cdek_response_for_tariff_code,
|
|
)
|
|
from app.adapters.delivery_providers.cdek.order_mapper import (
|
|
CDEKOrderInfo,
|
|
CDEKOrderMappingError,
|
|
CDEKOrderRegistrationResult,
|
|
CDEKWaybillInfo,
|
|
centimeters_string_to_int,
|
|
kilograms_string_to_grams,
|
|
map_cdek_existing_order_response,
|
|
map_cdek_order_info_response,
|
|
map_cdek_order_request,
|
|
map_cdek_order_response,
|
|
map_cdek_waybill_info_response,
|
|
resolve_cdek_city_code,
|
|
)
|
|
from app.cities import cities_map
|
|
from app.config import AdapterConfig
|
|
from app.schemas.payment import InitPaymentRequest
|
|
from app.schemas.request import DeliveryCalculationRequest
|
|
from app.schemas.response import DeliveryPrice
|
|
log = structlog.get_logger(__name__)
|
|
|
|
|
|
class CDEKClientError(ProviderClientError):
|
|
"""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._tariff_url = f"{normalized_base_url}/calculator/tarifflist"
|
|
self._orders_url = f"{normalized_base_url}/orders"
|
|
self._print_orders_url = f"{normalized_base_url}/print/orders"
|
|
self._timeout_seconds = timeout_seconds
|
|
self._retry_attempts = retry_attempts
|
|
self._retry_backoff_seconds = retry_backoff_seconds
|
|
self._sleep = sleep
|
|
|
|
async def get_raw_price(
|
|
self, request: DeliveryCalculationRequest
|
|
) -> dict[str, Any]:
|
|
payload = await self._build_payload(request)
|
|
return await self._post_tariff_payload(
|
|
payload,
|
|
request_error_message=None,
|
|
)
|
|
|
|
async def get_raw_payment_price(
|
|
self,
|
|
request: InitPaymentRequest,
|
|
) -> dict[str, Any]:
|
|
payload = self._build_payment_price_payload(request)
|
|
return await self._post_tariff_payload(
|
|
payload,
|
|
request_error_message=(
|
|
"CDEK payment price validation request was rejected with status"
|
|
),
|
|
)
|
|
|
|
async def _post_tariff_payload(
|
|
self,
|
|
payload: dict[str, Any],
|
|
*,
|
|
request_error_message: str | None,
|
|
) -> dict[str, Any]:
|
|
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
|
|
log.warning(
|
|
"cdek_tariff_request_server_error",
|
|
status_code=response.status_code,
|
|
response_body=_response_text_or_none(response),
|
|
request_payload=payload,
|
|
)
|
|
raise CDEKClientError(
|
|
f"CDEK tariff request failed with status {response.status_code}."
|
|
)
|
|
|
|
if request_error_message is not None and 400 <= response.status_code < 500:
|
|
log.warning(
|
|
"cdek_tariff_request_rejected",
|
|
status_code=response.status_code,
|
|
response_body=_response_text_or_none(response),
|
|
request_payload=payload,
|
|
)
|
|
raise CDEKRequestError(
|
|
f"{request_error_message} {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: InitPaymentRequest, order_uuid: str
|
|
) -> CDEKOrderRegistrationResult:
|
|
payload = map_cdek_order_request(request, order_uuid)
|
|
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:
|
|
response_payload = _response_json_or_none(response)
|
|
existing_result = map_cdek_existing_order_response(response_payload)
|
|
if existing_result is not None:
|
|
return existing_result
|
|
log.warning(
|
|
"cdek_order_register_rejected",
|
|
status_code=response.status_code,
|
|
request_payload=payload,
|
|
response_payload=response_payload,
|
|
response_body=_response_text_or_none(response),
|
|
)
|
|
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."
|
|
)
|
|
|
|
log.info(
|
|
"cdek_order_register_raw_response",
|
|
status_code=response.status_code,
|
|
response_payload=raw_payload,
|
|
)
|
|
|
|
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.")
|
|
|
|
async def get_order(self, cdek_order_uuid: str) -> CDEKOrderInfo:
|
|
url = f"{self._orders_url}/{cdek_order_uuid}"
|
|
raw_payload = await self._get_json(
|
|
url,
|
|
failure_message="CDEK get order failed",
|
|
)
|
|
requests_with_errors = _extract_requests_with_errors(raw_payload)
|
|
if requests_with_errors:
|
|
log.warning(
|
|
"cdek_order_request_errors",
|
|
cdek_order_uuid=cdek_order_uuid,
|
|
requests=requests_with_errors,
|
|
)
|
|
try:
|
|
return map_cdek_order_info_response(raw_payload)
|
|
except CDEKOrderMappingError as exc:
|
|
raise CDEKClientError(
|
|
"CDEK get order response payload is invalid."
|
|
) from exc
|
|
|
|
async def get_waybill(self, cdek_waybill_uuid: str) -> CDEKWaybillInfo:
|
|
url = f"{self._print_orders_url}/{cdek_waybill_uuid}"
|
|
raw_payload = await self._get_json(
|
|
url,
|
|
failure_message="CDEK get waybill failed",
|
|
)
|
|
try:
|
|
return map_cdek_waybill_info_response(raw_payload)
|
|
except CDEKOrderMappingError as exc:
|
|
raise CDEKClientError(
|
|
"CDEK get waybill response payload is invalid."
|
|
) from exc
|
|
|
|
async def download_waybill_pdf(self, url: str) -> bytes:
|
|
failure_message = "CDEK waybill PDF download failed"
|
|
for attempt in range(self._retry_attempts + 1):
|
|
try:
|
|
token = await self._auth_client.get_access_token()
|
|
response = await self._http_client.get(
|
|
url,
|
|
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(
|
|
f"{failure_message} 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"{failure_message} with retriable status "
|
|
f"{response.status_code}."
|
|
)
|
|
|
|
if 400 <= response.status_code < 500:
|
|
log.warning(
|
|
"cdek_waybill_pdf_download_rejected",
|
|
url=url,
|
|
status_code=response.status_code,
|
|
response_body=_response_text_or_none(response),
|
|
)
|
|
raise CDEKRequestError(
|
|
f"{failure_message} with status {response.status_code}."
|
|
)
|
|
|
|
try:
|
|
response.raise_for_status()
|
|
except httpx.HTTPError as exc:
|
|
raise CDEKClientError(
|
|
f"{failure_message}: invalid response."
|
|
) from exc
|
|
|
|
return response.content
|
|
|
|
raise CDEKClientError(f"{failure_message} unexpectedly.")
|
|
|
|
async def _get_json(self, url: str, *, failure_message: str) -> dict[str, Any]:
|
|
for attempt in range(self._retry_attempts + 1):
|
|
try:
|
|
token = await self._auth_client.get_access_token()
|
|
response = await self._http_client.get(
|
|
url,
|
|
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(
|
|
f"{failure_message} 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"{failure_message} with retriable status "
|
|
f"{response.status_code}."
|
|
)
|
|
|
|
if 400 <= response.status_code < 500:
|
|
log.warning(
|
|
"cdek_get_request_rejected",
|
|
url=url,
|
|
status_code=response.status_code,
|
|
response_body=_response_text_or_none(response),
|
|
)
|
|
raise CDEKRequestError(
|
|
f"{failure_message} with status {response.status_code}."
|
|
)
|
|
|
|
try:
|
|
response.raise_for_status()
|
|
raw_payload = response.json()
|
|
except (httpx.HTTPError, TypeError, ValueError) as exc:
|
|
raise CDEKClientError(
|
|
f"{failure_message}: invalid response payload."
|
|
) from exc
|
|
|
|
if not isinstance(raw_payload, dict):
|
|
raise CDEKClientError(
|
|
f"{failure_message}: payload must be a JSON object."
|
|
)
|
|
return raw_payload
|
|
|
|
raise CDEKClientError(f"{failure_message} 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: DeliveryCalculationRequest
|
|
) -> dict[str, Any]:
|
|
from_city_code = self._get_cdek_city_code(request.from_city)
|
|
to_city_code = self._get_cdek_city_code(request.to_city)
|
|
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)),
|
|
}
|
|
],
|
|
}
|
|
|
|
@staticmethod
|
|
def _build_payment_price_payload(
|
|
request: InitPaymentRequest,
|
|
) -> dict[str, Any]:
|
|
try:
|
|
from_code = resolve_cdek_city_code(request.sender_address.city_id)
|
|
to_code = resolve_cdek_city_code(request.receiver_address.city_id)
|
|
weight_grams = kilograms_string_to_grams(request.system_data.weight)
|
|
except CDEKOrderMappingError as exc:
|
|
raise CDEKRequestError(str(exc)) from exc
|
|
|
|
package: dict[str, Any] = {"weight": weight_grams}
|
|
dimensions = request.system_data.dimensions
|
|
if request.system_data.parcel_type == "parcel" and dimensions is not None:
|
|
try:
|
|
package["length"] = centimeters_string_to_int(dimensions.length, "length")
|
|
package["width"] = centimeters_string_to_int(dimensions.width, "width")
|
|
package["height"] = centimeters_string_to_int(dimensions.height, "height")
|
|
except CDEKOrderMappingError as exc:
|
|
raise CDEKRequestError(str(exc)) from exc
|
|
|
|
return {
|
|
"type": 2,
|
|
"from_location": {"code": from_code},
|
|
"to_location": {"code": to_code},
|
|
"packages": [package],
|
|
}
|
|
|
|
@staticmethod
|
|
def _get_cdek_city_code(city_id: int) -> int:
|
|
city_entry = cities_map.get(str(city_id))
|
|
if not isinstance(city_entry, dict):
|
|
raise CDEKRequestError(
|
|
f"CDEK city mapping is not configured for city id {city_id}."
|
|
)
|
|
|
|
cdek_data = city_entry.get("cdek")
|
|
if not isinstance(cdek_data, dict):
|
|
raise CDEKRequestError(
|
|
f"CDEK city mapping is not configured for city id {city_id}."
|
|
)
|
|
|
|
raw_city_code = cdek_data.get("code")
|
|
if raw_city_code is None or isinstance(raw_city_code, bool):
|
|
raise CDEKRequestError(
|
|
f"CDEK city code is invalid for city id {city_id}."
|
|
)
|
|
|
|
try:
|
|
return int(raw_city_code)
|
|
except (TypeError, ValueError) as exc:
|
|
raise CDEKRequestError(
|
|
f"CDEK city code is invalid for city id {city_id}."
|
|
) from exc
|
|
|
|
|
|
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_prices(
|
|
self, request: DeliveryCalculationRequest
|
|
) -> list[DeliveryPrice]:
|
|
raw_payload = await self._client.get_raw_price(request)
|
|
return map_cdek_response(raw_payload)
|
|
|
|
async def get_payment_price(
|
|
self,
|
|
request: InitPaymentRequest,
|
|
) -> DeliveryPrice | None:
|
|
raw_payload = await self._client.get_raw_payment_price(request)
|
|
try:
|
|
return map_cdek_response_for_tariff_code(
|
|
raw_payload,
|
|
tariff_code=request.system_data.tariff.tariff_code,
|
|
)
|
|
except CDEKMappingError as exc:
|
|
raise CDEKClientError(
|
|
"CDEK payment price validation response payload is invalid."
|
|
) from exc
|
|
|
|
async def register_order(
|
|
self, request: InitPaymentRequest, order_uuid: str
|
|
) -> CDEKOrderRegistrationResult:
|
|
return await self._client.register_order(request, order_uuid)
|
|
|
|
async def get_order(self, cdek_order_uuid: str) -> CDEKOrderInfo:
|
|
return await self._client.get_order(cdek_order_uuid)
|
|
|
|
async def get_waybill(self, cdek_waybill_uuid: str) -> CDEKWaybillInfo:
|
|
return await self._client.get_waybill(cdek_waybill_uuid)
|
|
|
|
|
|
def _response_json_or_none(response: httpx.Response) -> object | None:
|
|
try:
|
|
return response.json()
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _response_text_or_none(response: httpx.Response) -> str | None:
|
|
try:
|
|
return response.text
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _extract_requests_with_errors(
|
|
payload: dict[str, Any],
|
|
) -> list[dict[str, object]]:
|
|
requests = payload.get("requests")
|
|
if not isinstance(requests, list):
|
|
return []
|
|
|
|
requests_with_errors: list[dict[str, object]] = []
|
|
for request in requests:
|
|
if not isinstance(request, dict):
|
|
continue
|
|
errors = request.get("errors")
|
|
if not isinstance(errors, list) or not errors:
|
|
continue
|
|
requests_with_errors.append(
|
|
{
|
|
"request_uuid": request.get("request_uuid"),
|
|
"type": request.get("type"),
|
|
"state": request.get("state"),
|
|
"errors": errors,
|
|
}
|
|
)
|
|
return requests_with_errors
|