76 lines
3.3 KiB
Python
76 lines
3.3 KiB
Python
|
import json
|
||
|
|
import logging
|
||
|
|
from typing import Any, Dict, List, Optional
|
||
|
|
|
||
|
|
import aiohttp
|
||
|
|
|
||
|
|
from app.core.config import settings
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
class ShipmondoApiClient:
|
||
|
|
def __init__(self) -> None:
|
||
|
|
self.base_url = (settings.SHIPMONDO_API_BASE_URL or "").rstrip("/")
|
||
|
|
self.timeout_seconds = max(5, int(settings.SHIPMONDO_TIMEOUT_SECONDS or 30))
|
||
|
|
|
||
|
|
@property
|
||
|
|
def configured(self) -> bool:
|
||
|
|
return bool(self.base_url and settings.SHIPMONDO_API_USER and settings.SHIPMONDO_API_KEY)
|
||
|
|
|
||
|
|
async def _request(
|
||
|
|
self,
|
||
|
|
method: str,
|
||
|
|
path: str,
|
||
|
|
*,
|
||
|
|
params: Optional[Dict[str, Any]] = None,
|
||
|
|
json_payload: Optional[Dict[str, Any]] = None,
|
||
|
|
) -> Any:
|
||
|
|
if not self.configured:
|
||
|
|
raise RuntimeError("Shipmondo API credentials are not configured")
|
||
|
|
|
||
|
|
timeout = aiohttp.ClientTimeout(total=self.timeout_seconds)
|
||
|
|
auth = aiohttp.BasicAuth(settings.SHIPMONDO_API_USER, settings.SHIPMONDO_API_KEY)
|
||
|
|
headers = {"Accept": "application/json", "Content-Type": "application/json"}
|
||
|
|
url = f"{self.base_url}/{path.lstrip('/')}"
|
||
|
|
|
||
|
|
async with aiohttp.ClientSession(timeout=timeout, auth=auth, headers=headers) as session:
|
||
|
|
async with session.request(method, url, params=params, json=json_payload) as response:
|
||
|
|
body = await response.text()
|
||
|
|
if response.status >= 400:
|
||
|
|
detail = ""
|
||
|
|
try:
|
||
|
|
parsed = json.loads(body)
|
||
|
|
detail = str(parsed.get("message") or parsed.get("error") or parsed.get("errors") or "")
|
||
|
|
except (ValueError, AttributeError):
|
||
|
|
detail = body.strip()
|
||
|
|
detail = detail[:300]
|
||
|
|
logger.warning("Shipmondo request failed (%s %s): HTTP %s", method, path, response.status)
|
||
|
|
suffix = f": {detail}" if detail else ""
|
||
|
|
raise RuntimeError(f"Shipmondo API returned HTTP {response.status}{suffix}")
|
||
|
|
if not body.strip():
|
||
|
|
return {}
|
||
|
|
return json.loads(body)
|
||
|
|
|
||
|
|
async def list_products(self, country_code: str) -> List[Dict[str, Any]]:
|
||
|
|
payload = await self._request("GET", "/products", params={"country_code": country_code.upper()})
|
||
|
|
if isinstance(payload, list):
|
||
|
|
return [item for item in payload if isinstance(item, dict)]
|
||
|
|
if isinstance(payload, dict):
|
||
|
|
raw_items = payload.get("items") or payload.get("products") or payload.get("data") or []
|
||
|
|
if isinstance(raw_items, list):
|
||
|
|
return [item for item in raw_items if isinstance(item, dict)]
|
||
|
|
return []
|
||
|
|
|
||
|
|
async def create_shipment(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
||
|
|
response = await self._request("POST", "/shipments", json_payload=payload)
|
||
|
|
if not isinstance(response, dict):
|
||
|
|
raise RuntimeError("Shipmondo returned an invalid shipment response")
|
||
|
|
return response
|
||
|
|
|
||
|
|
async def get_shipment(self, shipment_id: str) -> Dict[str, Any]:
|
||
|
|
response = await self._request("GET", f"/shipments/{shipment_id}")
|
||
|
|
if not isinstance(response, dict):
|
||
|
|
raise RuntimeError("Shipmondo returned an invalid shipment response")
|
||
|
|
return response
|