"""Shared response and request models used across multiple endpoints."""
from __future__ import annotations
import logging
from typing import Generic, List, Optional, TypeVar
from pydantic import Field
from ab.api.models.base import ResponseModel
from ab.api.models.mixins import PaginatedRequestMixin, SortableRequestMixin
from ab.api.models.shipments import ShipmentWeight
logger = logging.getLogger(__name__)
T = TypeVar("T")
[docs]
class ServiceBaseResponse(ResponseModel):
"""Standard success/error wrapper returned by many ABConnect endpoints.
Supports boolean evaluation::
resp = api.some_endpoint(...)
if resp:
print("Success")
else:
resp.raise_for_error()
"""
success: Optional[bool] = Field(None, description="Whether the operation succeeded")
error_message: Optional[str] = Field(None, alias="errorMessage", description="Error detail when success is False")
job_sub_management_status: Optional[dict] = Field(
None, alias="jobSubManagementStatus", description="Current job sub-management status"
)
documents: Optional[List[str]] = Field(None, description="Document URLs/references generated by operation")
errors: Optional[dict] = Field(None, description="Detailed error object when success is False")
confirm_required: Optional[bool] = Field(
None, alias="confirmRequired", description="Whether user confirmation is needed"
)
notifications: Optional[List[str]] = Field(None, description="Notification messages for user")
shipment_id: Optional[str] = Field(None, alias="shipmentId", description="Shipment UUID created or modified")
shipment_accept_identifier: Optional[str] = Field(
None, alias="shipmentAcceptIdentifier", description="Carrier acceptance reference"
)
weight: Optional[ShipmentWeight] = Field(None, description="Shipment weight details")
total_net_charge_amount: Optional[float] = Field(
None, alias="totalNetChargeAmount", description="Total net charge amount"
)
currency_code: Optional[str] = Field(None, alias="currencyCode", description="Currency code (e.g., USD)")
international_info_required: Optional[bool] = Field(
None, alias="internationalInfoRequired", description="Whether international info is needed"
)
ship_out_date_required: Optional[bool] = Field(
None, alias="shipOutDateRequired", description="Whether ship-out date confirmation is needed"
)
fed_ex_express_freight_detail_required: Optional[bool] = Field(
None,
alias="fedExExpressFreightDetailRequired",
description="Whether FedEx Express freight details are required",
)
carrier_api: Optional[int] = Field(None, alias="carrierAPI", description="Carrier API type code")
[docs]
def raise_for_error(self) -> None:
"""Raise ``ValueError`` when ``success`` is ``False``."""
if self.success is False:
error_msg = self.error_message or "Unknown error"
logger.error("API request failed: %s", error_msg)
raise ValueError(error_msg)
def __bool__(self) -> bool:
return self.success is True
class ServiceWarningResponse(ServiceBaseResponse):
"""Extends :class:`ServiceBaseResponse` with an optional warning."""
warning_message: Optional[str] = Field(
None, alias="warningMessage", description="Warning present even on success"
)
def raise_for_error(self) -> None:
if self.warning_message:
logger.warning("API response warning: %s", self.warning_message)
super().raise_for_error()
[docs]
class PaginatedList(ResponseModel, Generic[T]):
"""Generic pagination wrapper used by the Catalog API.
Example::
result: PaginatedList[CatalogExpandedDto] = api.catalog.list(page_number=1)
for item in result.items:
print(item.title)
"""
items: List[T] = Field(default_factory=list, description="Page of results")
page_number: int = Field(0, alias="pageNumber", description="Current page (1-based)")
total_pages: int = Field(0, alias="totalPages", description="Total pages available")
total_items: int = Field(0, alias="totalItems", description="Total items across all pages")
has_previous_page: bool = Field(False, alias="hasPreviousPage")
has_next_page: bool = Field(False, alias="hasNextPage")
[docs]
class ListRequest(PaginatedRequestMixin, SortableRequestMixin):
"""Shared request body for paginated list endpoints (Companies, Users)."""
filters: Optional[dict] = Field(None, description="Filter criteria")