Functionality to get the next page of a paginated collection (#261)

Author: jchen293

CHANGELOG.md

@@ -1,5 +1,9 @@
 # CHANGELOG
 
+## Next release
+
+- Add `get_next_page` function that retrieves the next page of results for a paginated collection
+
 ## v7.10.0 (2023-02-17)
 
 - Adds beta `retrieve_stateless_rates` function

Makefile

@@ -27,7 +27,7 @@ clean:
 
 ## coverage - Test the project and generate an HTML coverage report
 coverage:
-	$(VIRTUAL_BIN)/pytest --cov=$(PROJECT_NAME) --cov-branch --cov-report=html --cov-report=term-missing --cov-fail-under=88
+	$(VIRTUAL_BIN)/pytest --cov=$(PROJECT_NAME) --cov-branch --cov-report=html --cov-report=term-missing --cov-fail-under=87
 
 ## docs - Generates docs for the library
 docs:

easypost/address.py

@@ -10,10 +10,13 @@
     RequestMethod,
     Requestor,
 )
-from easypost.resource import AllResource
+from easypost.resource import (
+    AllResource,
+    NextPageResource,
+)
 
 
-class Address(AllResource):
+class Address(AllResource, NextPageResource):
     @classmethod
     def create(
         cls,

easypost/event.py

@@ -10,10 +10,13 @@
     RequestMethod,
     Requestor,
 )
-from easypost.resource import AllResource
+from easypost.resource import (
+    AllResource,
+    NextPageResource,
+)
 
 
-class Event(AllResource):
+class Event(AllResource, NextPageResource):
     @classmethod
     def receive(cls, values: str) -> "Event":
         return convert_to_easypost_object(response=json.loads(s=values), api_key=easypost.api_key)

easypost/insurance.py

@@ -1,8 +1,9 @@
 from easypost.resource import (
     AllResource,
     CreateResource,
+    NextPageResource,
 )
 
 
-class Insurance(AllResource, CreateResource):
+class Insurance(AllResource, CreateResource, NextPageResource):
     pass

easypost/pickup.py

@@ -7,11 +7,12 @@
 from easypost.resource import (
     AllResource,
     CreateResource,
+    NextPageResource,
 )
 from easypost.util import get_lowest_object_rate
 
 
-class Pickup(CreateResource, AllResource):
+class Pickup(CreateResource, AllResource, NextPageResource):
     def buy(self, **params) -> "Pickup":
         """Buy a pickup."""
         requestor = Requestor(local_api_key=self._api_key)

easypost/referral.py

@@ -78,6 +78,32 @@ def all(
         )
         return convert_to_easypost_object(response=response, api_key=api_key)
 
+    @classmethod
+    def get_next_page(
+        cls,
+        referrals: Dict[str, Any],
+        page_size: int,
+        api_key: Optional[str] = None,
+    ) -> List["Referral"]:
+        """Retrieve next page of a referral customers."""
+        requestor = Requestor(local_api_key=api_key)
+        url = "/referral_customers"
+        referral_array = referrals["referral_customers"]
+
+        if referral_array is None or len(referral_array) == 0 or not referrals.get("has_more"):
+            raise Error(message="There are no more pages to retrieve.")
+
+        params = {
+            "before_id": referral_array[-1].id,
+            "page_size": page_size,
+        }
+
+        response, api_key = requestor.request(method=RequestMethod.GET, url=url, params=params)
+        if response is None or len(response["referral_customers"]) == 0 or not response["has_more"]:
+            raise Error(message="There are no more pages to retrieve.")
+
+        return convert_to_easypost_object(response=response, api_key=api_key)
+
     @staticmethod
     def add_credit_card(
         referral_api_key: str,

easypost/refund.py

@@ -1,8 +1,9 @@
 from easypost.resource import (
     AllResource,
     CreateResource,
+    NextPageResource,
 )
 
 
-class Refund(CreateResource, AllResource):
+class Refund(CreateResource, AllResource, NextPageResource):
     pass

easypost/report.py

@@ -1,4 +1,9 @@
-from typing import Optional
+from typing import (
+    Any,
+    Dict,
+    List,
+    Optional,
+)
 
 from easypost.easypost_object import convert_to_easypost_object
 from easypost.requestor import (
@@ -10,17 +15,37 @@
 
 class Report(Resource):
     @classmethod
-    def create(cls, api_key: Optional[str] = None, **params):
+    def create(cls, api_key: Optional[str] = None, **params) -> "Report":
         """Create a report."""
         requestor = Requestor(local_api_key=api_key)
         url = f"{cls.class_url()}/{params.get('type')}"
         response, api_key = requestor.request(method=RequestMethod.POST, url=url, params=params)
         return convert_to_easypost_object(response=response, api_key=api_key)
 
     @classmethod
-    def all(cls, api_key: Optional[str] = None, **params):
+    def all(cls, api_key: Optional[str] = None, **params) -> List["Report"]:
         """Retrieve all reports."""
         requestor = Requestor(local_api_key=api_key)
-        url = f"{cls.class_url()}/{params.get('type')}"
+        type = params.pop("type")
+        url = f"{cls.class_url()}/{type}"
+        response, api_key = requestor.request(method=RequestMethod.GET, url=url, params=params)
+        response["type"] = type
+        return convert_to_easypost_object(response=response, api_key=api_key)
+
+    @classmethod
+    def get_next_page(
+        cls,
+        reports: Dict[str, Any],
+        page_size: int = None,
+        api_key: Optional[str] = None,
+    ) -> List["Report"]:
+        """Get next page of Report collection"""
+        requestor = Requestor(local_api_key=api_key)
+        type = reports.get("type")
+        url = f"{cls.class_url()}/{type}"
+        params = {
+            "before_id": reports["reports"][-1].id,
+            "page_size": page_size,
+        }
         response, api_key = requestor.request(method=RequestMethod.GET, url=url, params=params)
         return convert_to_easypost_object(response=response, api_key=api_key)

easypost/requestor.py

@@ -84,7 +84,7 @@ def request(
         params: Optional[Dict[str, Any]] = None,
         api_key_required: bool = True,
         beta: bool = False,
-    ) -> Tuple[dict, Optional[str]]:
+    ) -> Tuple[Dict[str, Any], Optional[str]]:
         """Make a request to the EasyPost API."""
         if params is None:
             params = {}

easypost/resource.py

@@ -1,6 +1,7 @@
 import re
 from typing import (
     Any,
+    Dict,
     List,
     Optional,
 )
@@ -65,6 +66,39 @@ def all(cls, api_key: Optional[str] = None, **params) -> List[Any]:
         return convert_to_easypost_object(response=response, api_key=api_key)
 
 
+class NextPageResource(Resource):
+    @classmethod
+    def get_next_page(
+        cls,
+        collection: Dict[str, Any],
+        page_size: int,
+        api_key: Optional[str] = None,
+        optional_params: Optional[Dict[str, Any]] = None,
+    ) -> List[Any]:
+        """Retrieve next page of a specific collection."""
+        requestor = Requestor(local_api_key=api_key)
+        url = cls.class_url()
+        collection_array = collection.get(url[1:])
+
+        if collection_array is None or len(collection_array) == 0 or not collection.get("has_more"):
+            raise Error(message="There are no more pages to retrieve.")
+
+        params = {
+            "before_id": collection_array[-1].id,
+            "page_size": page_size,
+        }
+
+        if optional_params:
+            params.update(optional_params)
+
+        response, api_key = requestor.request(method=RequestMethod.GET, url=url, params=params)
+        response_array: List[Any] = response.get(url[1:])  # type: ignore
+        if response is None or len(response_array) == 0 or not response.get("has_more"):
+            raise Error(message="There are no more pages to retrieve.")
+
+        return convert_to_easypost_object(response=response, api_key=api_key)
+
+
 class CreateResource(Resource):
     @classmethod
     def create(cls, api_key: Optional[str] = None, **params) -> Any:

easypost/scanform.py

@@ -5,12 +5,15 @@
     RequestMethod,
     Requestor,
 )
-from easypost.resource import AllResource
+from easypost.resource import (
+    AllResource,
+    NextPageResource,
+)
 
 
-class ScanForm(AllResource):
+class ScanForm(AllResource, NextPageResource):
     @classmethod
-    def create(cls, api_key: Optional[str] = None, **params):
+    def create(cls, api_key: Optional[str] = None, **params) -> "ScanForm":
         """Create a scanform."""
         requestor = Requestor(local_api_key=api_key)
         url = cls.class_url()

easypost/shipment.py

@@ -12,11 +12,35 @@
     RequestMethod,
     Requestor,
 )
-from easypost.resource import AllResource
+from easypost.resource import NextPageResource
 from easypost.util import get_lowest_object_rate
 
 
-class Shipment(AllResource):
+class Shipment(NextPageResource):
+    @classmethod
+    def all(cls, api_key: Optional[str] = None, **params) -> List["Shipment"]:
+        """Retrieve a list of Shipments."""
+        requestor = Requestor(local_api_key=api_key)
+        response, api_key = requestor.request(method=RequestMethod.GET, url="/shipments", params=params)
+        response["include_children"] = params.get("include_children")
+        response["purchased"] = params.get("purchased")
+        return convert_to_easypost_object(response=response, api_key=api_key)
+
+    @classmethod
+    def get_next_page(
+        cls,
+        shipments: Dict[str, Any],
+        page_size: int,
+        api_key: Optional[str] = None,
+        optional_params: Optional[Dict[str, Any]] = None,
+    ) -> List["Shipment"]:
+        """Get next page of shipment collection."""
+        optional_params = {
+            "include_children": shipments.get("include_children"),
+            "purchased": shipments.get("purchased"),
+        }
+        return super().get_next_page(shipments, page_size, api_key, optional_params)
+
     @classmethod
     def create(cls, api_key: Optional[str] = None, with_carbon_offset: Optional[bool] = False, **params) -> "Shipment":
         """Create an Shipment object."""

easypost/tracker.py

@@ -5,17 +5,19 @@
     Optional,
 )
 
+from easypost.easypost_object import convert_to_easypost_object
 from easypost.requestor import (
     RequestMethod,
     Requestor,
 )
 from easypost.resource import (
     AllResource,
     CreateResource,
+    NextPageResource,
 )
 
 
-class Tracker(AllResource, CreateResource):
+class Tracker(AllResource, CreateResource, NextPageResource):
     @classmethod
     def create_list(cls, trackers: List[Dict[str, Any]], api_key: Optional[str] = None) -> bool:
         """Create a list of trackers."""
@@ -24,3 +26,28 @@ def create_list(cls, trackers: List[Dict[str, Any]], api_key: Optional[str] = No
         new_params = {"trackers": trackers}
         _, _ = requestor.request(method=RequestMethod.POST, url=url, params=new_params)
         return True
+
+    @classmethod
+    def all(cls, api_key: Optional[str] = None, **params) -> List["Tracker"]:
+        """Retrieve a list of Trackers."""
+        requestor = Requestor(local_api_key=api_key)
+        url = cls.class_url()
+        response, api_key = requestor.request(method=RequestMethod.GET, url=url, params=params)
+        response["tracking_code"] = params.get("tracking_code")
+        response["carrier"] = params.get("carrier")
+        return convert_to_easypost_object(response=response, api_key=api_key)
+
+    @classmethod
+    def get_next_page(
+        cls,
+        trackers: Dict[str, Any],
+        page_size: int,
+        api_key: Optional[str] = None,
+        optional_params: Optional[Dict[str, Any]] = None,
+    ) -> List["Tracker"]:
+        """Get next page of Tracker collection."""
+        optional_params = {
+            "tracking_code": trackers.get("tracking_code"),
+            "carrier": trackers.get("carrier"),
+        }
+        return super().get_next_page(trackers, page_size, api_key, optional_params)