"""Job-scoped shipment operations — swagger tag ``JobShipment`` (11 routes).
Exposed as ``api.jobs.shipment``. The legacy ``api.shipments``
endpoint retains the 3 non-job-scoped routes (``GET /shipment``,
``GET /shipment/accessorials``, ``GET /shipment/document/{docId}``); the
11 job-scoped routes move here and ``api.shipments`` keeps deprecation
shims that forward to this class.
Method renames (drop redundant ``_shipment`` suffix; standardise ``get_``):
* :meth:`get_rate_quotes` (unchanged)
* :meth:`request_rate_quotes` (unchanged)
* :meth:`book` (unchanged)
* :meth:`delete` (was ``delete_shipment``)
* :meth:`get_accessorials` (unchanged)
* :meth:`add_accessorial` (unchanged)
* :meth:`remove_accessorial` (unchanged)
* :meth:`get_origin_destination` (unchanged)
* :meth:`get_export_data` (unchanged)
* :meth:`post_export_data` (unchanged)
* :meth:`get_rates_state` (unchanged)
"""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ab.api.models.shared import ServiceBaseResponse
from ab.api.models.shipments import (
Accessorial,
AccessorialAddRequest,
JobCarrierRates,
RateQuote,
RatesState,
ShipmentBookRequest,
ShipmentExportData,
ShipmentExportRequest,
ShipmentOriginDestination,
ShipmentRateQuoteRequest,
)
from ab.api.base import BaseEndpoint
from ab.api.route import Route
_GET_RATE_QUOTES = Route(
"GET", "/job/{jobDisplayId}/shipment/ratequotes",
params_model="RateQuotesParams", response_model="List[RateQuote]",
)
_POST_RATE_QUOTES = Route(
"POST", "/job/{jobDisplayId}/shipment/ratequotes",
request_model="ShipmentRateQuoteRequest", response_model="List[RateQuote]",
)
_BOOK = Route(
"POST", "/job/{jobDisplayId}/shipment/book",
request_model="ShipmentBookRequest", response_model="ServiceBaseResponse",
)
_DELETE = Route("DELETE", "/job/{jobDisplayId}/shipment", response_model="ServiceBaseResponse")
_GET_ACCESSORIALS = Route(
"GET", "/job/{jobDisplayId}/shipment/accessorials", response_model="List[Accessorial]",
)
_ADD_ACCESSORIAL = Route(
"POST", "/job/{jobDisplayId}/shipment/accessorial",
request_model="AccessorialAddRequest", response_model="ServiceBaseResponse",
)
_REMOVE_ACCESSORIAL = Route(
"DELETE", "/job/{jobDisplayId}/shipment/accessorial/{addOnId}",
response_model="ServiceBaseResponse",
)
_GET_ORIGIN_DEST = Route(
"GET", "/job/{jobDisplayId}/shipment/origindestination",
response_model="ShipmentOriginDestination",
)
_GET_EXPORT_DATA = Route("GET", "/job/{jobDisplayId}/shipment/exportdata", response_model="ShipmentExportData")
_POST_EXPORT_DATA = Route(
"POST", "/job/{jobDisplayId}/shipment/exportdata",
request_model="ShipmentExportRequest", response_model="ServiceBaseResponse",
)
_GET_RATES_STATE = Route("GET", "/job/{jobDisplayId}/shipment/ratesstate", response_model="RatesState")
[docs]
class JobShipmentEndpoint(BaseEndpoint):
"""Job-scoped shipment operations (ACPortal API)."""
[docs]
def get_rate_quotes(self, job_display_id: int) -> list[RateQuote]:
"""``GET /job/{jobDisplayId}/shipment/ratequotes``
Compatibility list view over ``JobCarrierRates.rates``. Use
:meth:`get_rate_quotes_result` when you need ``ratesKey``,
``requestSnapshot``, or ``errors``.
Docs: https://ab-sdk.readthedocs.io/en/latest/api/jobs/shipment.get_rate_quotes.html
Query params: RateQuotesParams
Response model: List[RateQuote]
"""
return self._request(_GET_RATE_QUOTES.bind(jobDisplayId=job_display_id))
[docs]
def request_rate_quotes(self, job_display_id: int, *, data: ShipmentRateQuoteRequest | dict) -> list[RateQuote]:
"""``POST /job/{jobDisplayId}/shipment/ratequotes``
Compatibility list view over ``JobCarrierRates.rates``. Use
:meth:`request_rate_quotes_result` when you need ``ratesKey``,
``requestSnapshot``, or ``errors``.
Request model: :class:`ShipmentRateQuoteRequest`.
Docs: https://ab-sdk.readthedocs.io/en/latest/api/jobs/shipment.request_rate_quotes.html
Request model: ShipmentRateQuoteRequest
Response model: List[RateQuote]
"""
return self._request(_POST_RATE_QUOTES.bind(jobDisplayId=job_display_id), json=data)
[docs]
def get_rate_quotes_result(self, job_display_id: int) -> JobCarrierRates:
"""``GET /job/{jobDisplayId}/shipment/ratequotes``
Returns the full API 7.11 ``JobCarrierRates`` envelope, including the
``ratesKey`` needed by booking flows and any quote errors.
Response model: JobCarrierRates
"""
route = _GET_RATE_QUOTES.bind(jobDisplayId=job_display_id)
response = self._client.request(route.method, route.path)
return self._coerce_rate_quotes_result(response)
[docs]
def request_rate_quotes_result(
self,
job_display_id: int,
*,
data: ShipmentRateQuoteRequest | dict,
) -> JobCarrierRates:
"""``POST /job/{jobDisplayId}/shipment/ratequotes``
Returns the full API 7.11 ``JobCarrierRates`` envelope, including the
``ratesKey`` needed by booking flows and any quote errors.
Request model: :class:`ShipmentRateQuoteRequest`.
Request model: ShipmentRateQuoteRequest
Response model: JobCarrierRates
"""
payload = self._resolve_model(_POST_RATE_QUOTES.request_model).check(data)
route = _POST_RATE_QUOTES.bind(jobDisplayId=job_display_id)
response = self._client.request(route.method, route.path, json=payload)
return self._coerce_rate_quotes_result(response)
@staticmethod
def _coerce_rate_quotes_result(response: object) -> JobCarrierRates:
from ab.api.models.shipments import JobCarrierRates
if response is None:
return JobCarrierRates()
if isinstance(response, list):
return JobCarrierRates(rates=response)
return JobCarrierRates.model_validate(response)
[docs]
def book(self, job_display_id: int, *, data: ShipmentBookRequest | dict) -> ServiceBaseResponse:
"""``POST /job/{jobDisplayId}/shipment/book``
Request model: :class:`ShipmentBookRequest`.
Docs: https://ab-sdk.readthedocs.io/en/latest/api/jobs/shipment.book.html
Request model: ShipmentBookRequest
Response model: ServiceBaseResponse
"""
return self._request(_BOOK.bind(jobDisplayId=job_display_id), json=data)
[docs]
def delete(self, job_display_id: int) -> ServiceBaseResponse:
"""``DELETE /job/{jobDisplayId}/shipment``
Docs: https://ab-sdk.readthedocs.io/en/latest/api/jobs/shipment.delete.html
Response model: ServiceBaseResponse
"""
return self._request(_DELETE.bind(jobDisplayId=job_display_id))
[docs]
def get_accessorials(self, job_display_id: int) -> list[Accessorial]:
"""``GET /job/{jobDisplayId}/shipment/accessorials``
Docs: https://ab-sdk.readthedocs.io/en/latest/api/jobs/shipment.get_accessorials.html
Response model: List[Accessorial]
"""
return self._request(_GET_ACCESSORIALS.bind(jobDisplayId=job_display_id))
[docs]
def add_accessorial(self, job_display_id: int, *, data: AccessorialAddRequest | dict) -> ServiceBaseResponse:
"""``POST /job/{jobDisplayId}/shipment/accessorial``
Request model: :class:`AccessorialAddRequest`.
Docs: https://ab-sdk.readthedocs.io/en/latest/api/jobs/shipment.add_accessorial.html
Request model: AccessorialAddRequest
Response model: ServiceBaseResponse
"""
return self._request(_ADD_ACCESSORIAL.bind(jobDisplayId=job_display_id), json=data)
[docs]
def remove_accessorial(self, job_display_id: int, add_on_id: str) -> ServiceBaseResponse:
"""``DELETE /job/{jobDisplayId}/shipment/accessorial/{addOnId}``
Docs: https://ab-sdk.readthedocs.io/en/latest/api/jobs/shipment.remove_accessorial.html
Response model: ServiceBaseResponse
"""
return self._request(
_REMOVE_ACCESSORIAL.bind(jobDisplayId=job_display_id, addOnId=add_on_id),
)
[docs]
def get_origin_destination(self, job_display_id: int) -> ShipmentOriginDestination:
"""``GET /job/{jobDisplayId}/shipment/origindestination``
Docs: https://ab-sdk.readthedocs.io/en/latest/api/jobs/shipment.get_origin_destination.html
Response model: ShipmentOriginDestination
"""
return self._request(_GET_ORIGIN_DEST.bind(jobDisplayId=job_display_id))
[docs]
def get_export_data(self, job_display_id: int) -> ShipmentExportData:
"""``GET /job/{jobDisplayId}/shipment/exportdata``
Docs: https://ab-sdk.readthedocs.io/en/latest/api/jobs/shipment.get_export_data.html
Response model: ShipmentExportData
"""
return self._request(_GET_EXPORT_DATA.bind(jobDisplayId=job_display_id))
[docs]
def post_export_data(self, job_display_id: int, *, data: ShipmentExportRequest | dict) -> ServiceBaseResponse:
"""``POST /job/{jobDisplayId}/shipment/exportdata``
The body is the portal's flat ``InternationalParams`` shape —
:class:`ShipmentExportRequest`, ``customsValue`` required. GET returns
a ``jobItemId`` on each commodity that the POST schema forbids
(``additionalProperties: false``), so it is stripped here to keep the
GET → edit → POST round-trip legal.
Docs: https://ab-sdk.readthedocs.io/en/latest/api/jobs/shipment.post_export_data.html
Request model: ShipmentExportRequest
Response model: ServiceBaseResponse
"""
payload = self._resolve_model(_POST_EXPORT_DATA.request_model).check(data)
for commodity in payload.get("commodities") or []:
if isinstance(commodity, dict):
commodity.pop("jobItemId", None)
return self._request(_POST_EXPORT_DATA.bind(jobDisplayId=job_display_id), json=payload)
[docs]
def get_rates_state(self, job_display_id: int) -> RatesState:
"""``GET /job/{jobDisplayId}/shipment/ratesstate``
Docs: https://ab-sdk.readthedocs.io/en/latest/api/jobs/shipment.get_rates_state.html
Response model: RatesState
"""
return self._request(_GET_RATES_STATE.bind(jobDisplayId=job_display_id))