146 lines
5.0 KiB
Python
146 lines
5.0 KiB
Python
"""Cliente mínimo para Packlink PRO.
|
|
|
|
A Packlink PRO usa API key no header Authorization, sem Bearer.
|
|
Base testada: https://api.packlink.com/v1
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import httpx
|
|
|
|
from app.config import settings
|
|
|
|
|
|
class PacklinkError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PacklinkConfig:
|
|
base_url: str
|
|
api_key: str
|
|
|
|
|
|
def get_packlink_config() -> PacklinkConfig:
|
|
base_url = (settings.packlink_base_url or "https://api.packlink.com/v1").rstrip("/")
|
|
api_key = (settings.packlink_api_key or "").strip()
|
|
if not api_key:
|
|
raise PacklinkError("PACKLINK_API_KEY em falta")
|
|
return PacklinkConfig(base_url=base_url, api_key=api_key)
|
|
|
|
|
|
class PacklinkClient:
|
|
def __init__(self, config: Optional[PacklinkConfig] = None, timeout: float = 30.0):
|
|
self.config = config or get_packlink_config()
|
|
self.timeout = timeout
|
|
|
|
@property
|
|
def headers(self) -> Dict[str, str]:
|
|
return {
|
|
"Authorization": self.config.api_key,
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
async def _request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
params: Optional[Dict[str, Any]] = None,
|
|
json_body: Optional[Dict[str, Any]] = None,
|
|
) -> Any:
|
|
url = f"{self.config.base_url}/{path.lstrip('/')}"
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.request(
|
|
method.upper(),
|
|
url,
|
|
headers=self.headers,
|
|
params=params,
|
|
json=json_body,
|
|
)
|
|
|
|
if response.status_code >= 400:
|
|
detail = response.text
|
|
try:
|
|
detail = response.json()
|
|
except Exception:
|
|
pass
|
|
raise PacklinkError(f"Packlink error {response.status_code} em {method.upper()} {path}: {detail}")
|
|
|
|
if not response.content:
|
|
return None
|
|
try:
|
|
return response.json()
|
|
except Exception as exc:
|
|
raise PacklinkError(f"Resposta Packlink não é JSON em {method.upper()} {path}: {response.text[:500]}") from exc
|
|
|
|
async def get_client(self) -> Dict[str, Any]:
|
|
return await self._request("GET", "/clients")
|
|
|
|
async def get_warehouses(self) -> List[Dict[str, Any]]:
|
|
data = await self._request("GET", "/clients/warehouses")
|
|
return data or []
|
|
|
|
async def get_parcels(self) -> List[Dict[str, Any]]:
|
|
data = await self._request("GET", "/users/parcels")
|
|
return data or []
|
|
|
|
async def validate_postal_code(self, country: str, zip_code: str) -> List[Dict[str, Any]]:
|
|
return await self._request("GET", f"/locations/postalcodes/{country}/{zip_code}")
|
|
|
|
async def quote_services(
|
|
self,
|
|
*,
|
|
from_country: str,
|
|
from_zip: str,
|
|
to_country: str,
|
|
to_zip: str,
|
|
packages: List[Dict[str, Any]],
|
|
source: str = "PRO",
|
|
) -> List[Dict[str, Any]]:
|
|
params: Dict[str, Any] = {
|
|
"from[country]": from_country,
|
|
"from[zip]": from_zip,
|
|
"to[country]": to_country,
|
|
"to[zip]": to_zip,
|
|
"source": source,
|
|
}
|
|
for idx, package in enumerate(packages):
|
|
params[f"packages[{idx}][height]"] = package.get("height")
|
|
params[f"packages[{idx}][width]"] = package.get("width")
|
|
params[f"packages[{idx}][length]"] = package.get("length")
|
|
params[f"packages[{idx}][weight]"] = package.get("weight")
|
|
data = await self._request("GET", "/services", params=params)
|
|
return data or []
|
|
|
|
async def get_service_details(self, service_id: str | int) -> Dict[str, Any]:
|
|
return await self._request("GET", f"/services/available/{service_id}/details")
|
|
|
|
async def create_shipment(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
return await self._request("POST", "/shipments", json_body=payload)
|
|
|
|
async def list_shipments(self, *, limit: int = 100, offset: int = 0) -> Any:
|
|
"""List shipments when Packlink exposes the collection endpoint.
|
|
|
|
Some Packlink tenants/API versions may restrict this endpoint. The
|
|
reconciliation sync catches those API errors and reports them instead
|
|
of creating operational tasks.
|
|
"""
|
|
params = {"limit": min(max(int(limit), 1), 100), "offset": max(int(offset), 0)}
|
|
return await self._request("GET", "/shipments", params=params)
|
|
|
|
async def get_shipment(self, reference: str) -> Dict[str, Any]:
|
|
return await self._request("GET", f"/shipments/{reference}")
|
|
|
|
async def get_labels(self, reference: str) -> List[Any]:
|
|
data = await self._request("GET", f"/shipments/{reference}/labels")
|
|
return data or []
|
|
|
|
async def get_tracking(self, reference: str) -> List[Dict[str, Any]]:
|
|
data = await self._request("GET", f"/shipments/{reference}/track")
|
|
return data or []
|