[feat] Add claims API service

CHANGELOG.md

@@ -1,5 +1,9 @@
 # CHANGELOG
 
+## Next Release
+
+- Adds new `Claim` service for filing claims on EasyPost shipments and insurances
+
 ## v9.3.0 (2024-07-12)
 
 - Adds new `shipment.recommend_ship_date`, `smartrate.recommend_ship_date`, and `smartrate.estimate_delivery_date` functions

easypost/easypost_client.py

@@ -18,6 +18,7 @@
     BillingService,
     CarrierAccountService,
     CarrierMetadataService,
+    ClaimService,
     CustomsInfoService,
     CustomsItemService,
     EndShipperService,
@@ -62,6 +63,7 @@ def __init__(
         self.billing = BillingService(self)
         self.carrier_account = CarrierAccountService(self)
         self.carrier_metadata = CarrierMetadataService(self)
+        self.claim = ClaimService(self)
         self.customs_info = CustomsInfoService(self)
         self.customs_item = CustomsItemService(self)
         self.end_shipper = EndShipperService(self)

easypost/easypost_object.py

@@ -18,6 +18,7 @@
     "brd": "Brand",
     "ca": "CarrierAccount",
     "cfrep": "Report",
+    "clm": "Claim",
     "cstinfo": "CustomsInfo",
     "cstitem": "CustomsItem",
     "es": "EndShipper",

easypost/models/__init__.py

@@ -5,6 +5,7 @@
 from easypost.models.billing import Billing
 from easypost.models.brand import Brand
 from easypost.models.carrier_account import CarrierAccount
+from easypost.models.claim import Claim
 from easypost.models.customs_info import CustomsInfo
 from easypost.models.customs_item import CustomsItem
 from easypost.models.end_shipper import EndShipper

easypost/models/claim.py

@@ -0,0 +1,5 @@
+from easypost.easypost_object import EasyPostObject
+
+
+class Claim(EasyPostObject):
+    pass

easypost/services/__init__.py

@@ -7,6 +7,7 @@
 from easypost.services.billing_service import BillingService
 from easypost.services.carrier_account_service import CarrierAccountService
 from easypost.services.carrier_metadata_service import CarrierMetadataService
+from easypost.services.claim_service import ClaimService
 from easypost.services.customs_info_service import CustomsInfoService
 from easypost.services.customs_item_service import CustomsItemService
 from easypost.services.end_shipper_service import EndShipperService

easypost/services/base_service.py

@@ -42,47 +42,51 @@ def _instance_url(self, class_name: str, id: str) -> str:
         """Generate an instance URL based on a class name and ID."""
         return f"{self._class_url(class_name)}/{id}"
 
-    def _create_resource(self, class_name: str, **params) -> Any:
+    def _create_resource(self, class_name: str, beta: bool = False, **params) -> Any:
         """Create an EasyPost object via the EasyPost API."""
         url = self._class_url(class_name)
         wrapped_params = {self._snakecase_name(class_name): params}
 
-        response = Requestor(self._client).request(method=RequestMethod.POST, url=url, params=wrapped_params)
+        response = Requestor(self._client).request(method=RequestMethod.POST, url=url, params=wrapped_params, beta=beta)
 
         return convert_to_easypost_object(response=response)
 
-    def _all_resources(self, class_name: str, filters: Optional[Dict[str, Any]] = None, **params) -> Any:
+    def _all_resources(
+        self, class_name: str, filters: Optional[Dict[str, Any]] = None, beta: bool = False, **params
+    ) -> Any:
         """Retrieve a list of EasyPostObjects from the EasyPost API."""
         url = self._class_url(class_name)
-        response = Requestor(self._client).request(method=RequestMethod.GET, url=url, params=params)
+        response = Requestor(self._client).request(method=RequestMethod.GET, url=url, params=params, beta=beta)
 
         if filters:  # presence of filters indicates we are dealing with a paginated response
             response[_FILTERS_KEY] = filters  # Save the filters used to reference in potential get_next_page call
 
         return convert_to_easypost_object(response=response)
 
-    def _retrieve_resource(self, class_name: str, id: str) -> Any:
+    def _retrieve_resource(self, class_name: str, id: str, beta: bool = False) -> Any:
         """Retrieve an object from the EasyPost API."""
         url = self._instance_url(class_name, id)
 
-        response = Requestor(self._client).request(method=RequestMethod.GET, url=url)
+        response = Requestor(self._client).request(method=RequestMethod.GET, url=url, beta=beta)
 
         return convert_to_easypost_object(response=response)
 
-    def _update_resource(self, class_name: str, id: str, method: RequestMethod = RequestMethod.PATCH, **params) -> Any:
+    def _update_resource(
+        self, class_name: str, id: str, method: RequestMethod = RequestMethod.PATCH, beta: bool = False, **params
+    ) -> Any:
         """Update an EasyPost object via the EasyPost API."""
         url = self._instance_url(class_name, id)
         wrapped_params = {self._snakecase_name(class_name): params}
 
-        response = Requestor(self._client).request(method=method, url=url, params=wrapped_params)
+        response = Requestor(self._client).request(method=method, url=url, params=wrapped_params, beta=beta)
 
         return convert_to_easypost_object(response=response)
 
-    def _delete_resource(self, class_name: str, id: str) -> Any:
+    def _delete_resource(self, class_name: str, id: str, beta: bool = False) -> Any:
         """Delete an EasyPost object via the EasyPost API."""
         url = self._instance_url(class_name, id)
 
-        response = Requestor(self._client).request(method=RequestMethod.DELETE, url=url)
+        response = Requestor(self._client).request(method=RequestMethod.DELETE, url=url, beta=beta)
 
         return convert_to_easypost_object(response=response)
 

easypost/services/claim_service.py

@@ -0,0 +1,70 @@
+from typing import (
+    Any,
+    Dict,
+    Optional,
+)
+
+from easypost.easypost_object import convert_to_easypost_object
+from easypost.models import Claim
+from easypost.requestor import (
+    RequestMethod,
+    Requestor,
+)
+from easypost.services.base_service import BaseService
+
+
+class ClaimService(BaseService):
+    def __init__(self, client):
+        self._client = client
+        self._model_class = Claim.__name__
+
+    def create(self, **params) -> Claim:
+        """Create a Claim."""
+        url = "/claims"
+
+        response = Requestor(self._client).request(method=RequestMethod.POST, url=url, params=params, beta=False)
+
+        return convert_to_easypost_object(response=response)
+
+    def all(self, **params) -> Dict[str, Any]:
+        """Retrieve a list of Claims."""
+        filters = {
+            "key": "claims",
+        }
+
+        return self._all_resources(class_name=self._model_class, filters=filters, beta=False, **params)
+
+    def retrieve(self, id: str) -> Claim:
+        """Retrieve a Claim."""
+        return self._retrieve_resource(class_name=self._model_class, id=id, beta=False)
+
+    def get_next_page(
+        self,
+        claims: Dict[str, Any],
+        page_size: int,
+        optional_params: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """Retrieve the next page of the list Claim response."""
+        self._check_has_next_page(collection=claims)
+
+        params = {
+            "before_id": claims["claims"][-1].id,
+            "page_size": page_size,
+        }
+
+        if optional_params:
+            params.update(optional_params)
+
+        return self.all(**params)
+
+    def cancel(self, id: str) -> Claim:
+        """Cancel a Claim."""
+        url = f"/claims/{id}/cancel"
+
+        response = Requestor(self._client).request(
+            method=RequestMethod.POST,
+            url=url,
+            beta=False,
+        )
+
+        return convert_to_easypost_object(response=response)