Add Brenger API v2 client and models

brenger/client.py

@@ -1,16 +1,32 @@
 import logging
 
 import requests
-from requests import Response
+from requests import RequestException, Response
 
 from .exceptions import APIClientError, APIServerError
-from .models import ShipmentCreateRequest, ShipmentResponse
+from .models import (ShipmentCreateRequest, ShipmentResponse, V2QuoteRequest,
+                     V2QuoteResponse, V2RefundResponse,
+                     V2ShipmentCreateRequest, V2ShipmentCreateResponse,
+                     V2StatusResponse)
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 BASE_URL = "https://external-api.brenger.nl/{namespace}"
 HEADERS = {"Accept": "application/json", "Content-Type": "application/json"}
+DEFAULT_TIMEOUT = 30
+
+
+def _extract_error_details(response: Response) -> tuple[str, str, str]:
+    try:
+        response_json = response.json()
+    except ValueError:
+        return ("An error occurred", "Hint not provided", "Validation not provided")
+
+    error_description = response_json.get("description", "An error occurred")
+    error_hint = response_json.get("hint", "Hint not provided")
+    error_validation = response_json.get("validation_errors", "Validation not provided")
+    return (error_description, error_hint, error_validation)
 
 
 class BrengerAPIClient:
@@ -39,22 +55,107 @@ def get_shipment(self, shipment_id: str) -> ShipmentResponse:
         return ShipmentResponse(**response.json())
 
     def _handle_response_errors(self, response: Response) -> None:
-        if response.status_code == 500:
-            logger.error("Server Error")
-            raise APIServerError("Internal Server Error")
-        elif response.status_code >= 400:
-            response_json = response.json()
-            error_description = response_json.get("description", "An error occurred")
-            error_hint = response_json.get("hint", "Hint not provided")
-            error_validation = response_json.get(
-                "validation_errors", "Validation not provided"
+        if 500 <= response.status_code < 600:
+            logger.error("Server Error: status code %s", response.status_code)
+            raise APIServerError(
+                f"Brenger API server error (status code: {response.status_code})"
+            )
+        if response.status_code >= 400:
+            error_description, error_hint, error_validation = _extract_error_details(
+                response
+            )
+            error_message = (
+                f" Status code: {response.status_code} - "
+                f"Error description: {error_description} -"
+                f" Error hint: {error_hint} - "
+                f" validation errors: {error_validation}"
+            )
+            logger.error("Client Error: %s", error_message)
+            raise APIClientError(f"Client Error: {error_message}")
+
+
+V2_BASE_URL = "https://external-api.brenger.nl/v2/partners"
+
+
+class BrengerV2APIClient:
+    def __init__(
+        self, api_key: str, timeout: int = DEFAULT_TIMEOUT, base_url: str = None
+    ) -> None:
+        self.api_key = api_key
+        self.timeout = timeout
+        self.base_url = base_url or V2_BASE_URL
+        self.session = requests.Session()
+        self.session.headers.update({"X-AUTH-TOKEN": self.api_key, **HEADERS})
+
+    def get_quote(self, quote_data: V2QuoteRequest) -> V2QuoteResponse:
+        url = f"{self.base_url}/quote"
+        response = self._request(
+            "post",
+            url,
+            data=quote_data.model_dump_json(exclude_none=True).encode("utf-8"),
+        )
+        self._handle_response_errors(response)
+        logger.info("Quote retrieved successfully")
+        return V2QuoteResponse(**response.json())
+
+    def create_shipment(
+        self, shipment_data: V2ShipmentCreateRequest
+    ) -> V2ShipmentCreateResponse:
+        url = f"{self.base_url}/shipments"
+        response = self._request(
+            "post",
+            url,
+            data=shipment_data.model_dump_json(exclude_none=True).encode("utf-8"),
+        )
+        self._handle_response_errors(response)
+        logger.info(
+            "V2 Shipment created successfully with ID: %s",
+            response.json().get("shipment_id"),
+        )
+        return V2ShipmentCreateResponse(**response.json())
+
+    def get_shipment_status(self, shipment_id: str) -> V2StatusResponse:
+        url = f"{self.base_url}/shipments/{shipment_id}/status"
+        response = self._request("get", url)
+        self._handle_response_errors(response)
+        logger.info("Shipment status retrieved for ID: %s", shipment_id)
+        return V2StatusResponse(**response.json())
+
+    def cancel_shipment(self, shipment_id: str) -> None:
+        url = f"{self.base_url}/shipments/{shipment_id}/cancel"
+        response = self._request("post", url)
+        self._handle_response_errors(response)
+        logger.info("Shipment cancelled successfully for ID: %s", shipment_id)
+
+    def get_refund(self, shipment_id: str) -> V2RefundResponse:
+        url = f"{self.base_url}/shipments/{shipment_id}/refunds"
+        response = self._request("get", url)
+        self._handle_response_errors(response)
+        logger.info("Refund retrieved for shipment ID: %s", shipment_id)
+        return V2RefundResponse(**response.json())
+
+    def _request(self, method: str, url: str, **kwargs) -> Response:
+        try:
+            return self.session.request(method, url, timeout=self.timeout, **kwargs)
+        except RequestException as exc:
+            logger.error("Network error while calling Brenger v2 API", exc_info=True)
+            raise APIServerError(f"Failed to call Brenger v2 API: {exc}") from exc
+
+    def _handle_response_errors(self, response: Response) -> None:
+        if 500 <= response.status_code < 600:
+            logger.error("Server Error: status code %s", response.status_code)
+            raise APIServerError(
+                f"Brenger API server error (status code: {response.status_code})"
+            )
+        if response.status_code >= 400:
+            error_description, error_hint, error_validation = _extract_error_details(
+                response
             )
             error_message = (
                 f" Status code: {response.status_code} - "
                 f"Error description: {error_description} -"
                 f" Error hint: {error_hint} - "
                 f" validation errors: {error_validation}"
             )
-            logger.error(f"Client Error: {error_message}")
+            logger.error("Client Error: %s", error_message)
             raise APIClientError(f"Client Error: {error_message}")
-        response.raise_for_status()

brenger/models.py

@@ -145,3 +145,151 @@ class ShipmentResponse(BaseModel):
 
     class Config:
         extra = "ignore"
+
+
+# ============================================================================
+# V2 API Models
+# ============================================================================
+
+
+class V2Address(BaseModel):
+    country: str
+    administrative_area: Optional[str] = None
+    locality: str
+    postal_code: str
+    line1: str
+    line2: Optional[str] = None
+    lat: Optional[float] = None
+    lng: Optional[float] = None
+
+    _strip_whitespace = field_validator(
+        "line1", "line2", "postal_code", "locality", mode="before"
+    )(strip_whitespace)
+
+
+class V2Stop(BaseModel):
+    email: str
+    phone_number: Optional[str] = None
+    instructions: Optional[str] = None
+    first_name: str
+    last_name: Optional[str] = None
+    company_name: Optional[str] = None
+    preferred_locale: Optional[str] = None
+    address: V2Address
+
+    _strip_whitespace = field_validator(
+        "email", "first_name", "last_name", "company_name", mode="before"
+    )(strip_whitespace)
+
+
+class V2Item(BaseModel):
+    title: str
+    category: str
+    images: Optional[List[str]] = None
+    width: int
+    height: int
+    length: int
+    count: int
+
+    _strip_whitespace = field_validator("title", "category", mode="before")(
+        strip_whitespace
+    )
+
+
+class V2Amount(BaseModel):
+    currency: str
+    value: str
+
+
+class V2Price(BaseModel):
+    vat: V2Amount
+    incl_vat: V2Amount
+    excl_vat: V2Amount
+
+
+class V2Feasible(BaseModel):
+    value: bool
+    reasons: List[str]
+
+
+class V2AddressWrapper(BaseModel):
+    address: V2Address
+
+
+class V2QuoteRequest(BaseModel):
+    pickup: V2AddressWrapper
+    delivery: V2AddressWrapper
+    external_reference: Optional[str] = None
+    items: List[V2Item]
+
+
+class V2QuoteResponse(BaseModel):
+    price: V2Price
+    feasible: V2Feasible
+    pickup: V2AddressWrapper
+    delivery: V2AddressWrapper
+    external_reference: Optional[str] = None
+    items: List[V2Item]
+
+    class Config:
+        extra = "ignore"
+
+
+class V2ShipmentCreateRequest(BaseModel):
+    pickup: V2Stop
+    delivery: V2Stop
+    external_reference: str
+    external_listing_url: Optional[str] = None
+    external_private_url: Optional[str] = None
+    items: List[V2Item]
+    price: V2Price
+
+
+class V2ShipmentCreateResponse(BaseModel):
+    shipment_id: str
+    pickup: V2Stop
+    delivery: V2Stop
+    pickup_url: str
+    delivery_url: str
+    shipment_url: str
+    external_reference: str
+    external_listing_url: Optional[str] = None
+    external_private_url: Optional[str] = None
+    items: List[V2Item]
+    price: V2Price
+
+    class Config:
+        extra = "ignore"
+
+
+class V2Event(BaseModel):
+    id: str
+    timestamp: str
+    status: str
+
+
+class V2StatusResponse(BaseModel):
+    shipment_id: str
+    external_reference: str
+    status: str
+    events: List[V2Event]
+
+    class Config:
+        extra = "ignore"
+
+
+class V2RefundResponse(BaseModel):
+    refund_id: str
+    amount: V2Price
+    code: str
+
+    class Config:
+        extra = "ignore"
+
+
+class V2WebhookPayload(BaseModel):
+    event_id: str
+    shipment_id: str
+    external_reference: str
+    timestamp: str
+    status: str

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "brenger"
-version = "0.1.8"
+version = "2.0.0"
 description = ""
 authors = ["Omer <omer@whoppah.com>"]
 readme = "README.md"