feat: Add seam.create_paginator

README.rst

@@ -47,6 +47,16 @@ Contents
 
   * `Action Attempts <Action Attempts_>`_
 
+  * `Pagination <Pagination_>`_
+
+    * `Manually fetch pages with the nextPageCursor <Manually fetch pages with the nextPageCursor_>`_
+
+    * `Resume pagination <Resume pagination_>`_
+
+    * `Iterate over all resources <Iterate over all resources_>`_
+
+    * `Return all resources across all pages as a list <Return all resources across all pages as a list_>`_
+
   * `Interacting with Multiple Workspaces <Interacting with Multiple Workspaces_>`_
 
   * `Webhooks <Webhooks_>`_
@@ -257,6 +267,102 @@ For example:
   except SeamActionAttemptTimeoutError as e:
       print("Door took too long to unlock")
 
+Pagination
+~~~~~~~~~~
+
+Some Seam API endpoints that return lists of resources support pagination.
+Use the ``SeamPaginator`` class to fetch and process resources across multiple pages.
+
+Manually fetch pages with the nextPageCursor
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+.. code-block:: python
+
+  from seam import Seam
+
+  seam = Seam()
+
+  paginator = seam.create_paginator(seam.connected_accounts.list, {"limit": 20})
+
+  connected_accounts, pagination = paginator.first_page()
+
+  if pagination.has_next_page:
+      more_connected_accounts, _ = paginator.next_page(pagination.next_page_cursor)
+
+Resume pagination
+^^^^^^^^^^^^^^^^^
+
+Get the first page on initial load and store the state (e.g., in memory or a file):
+
+.. code-block:: python
+
+  import json
+  from seam import Seam
+
+  seam = Seam()
+
+  params = {"limit": 20}
+  paginator = seam.create_paginator(seam.connected_accounts.list, params)
+
+  connected_accounts, pagination = paginator.first_page()
+
+  # Example: Store state for later use (e.g., in a file or database)
+  pagination_state = {
+      "params": params,
+      "next_page_cursor": pagination.next_page_cursor,
+      "has_next_page": pagination.has_next_page,
+  }
+  with open("/tmp/seam_connected_accounts_list.json", "w") as f:
+      json.dump(pagination_state, f)
+
+Get the next page at a later time using the stored state:
+
+.. code-block:: python
+
+  import json
+  from seam import Seam
+
+  seam = Seam()
+
+  # Example: Load state from where it was stored
+  with open("/tmp/seam_connected_accounts_list.json", "r") as f:
+      pagination_state = json.load(f)
+
+  if pagination_state.get("has_next_page"):
+      paginator = seam.create_paginator(
+          seam.connected_accounts.list, pagination_state["params"]
+      )
+      more_connected_accounts, _ = paginator.next_page(
+          pagination_state["next_page_cursor"]
+      )
+
+Iterate over all resources
+^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+.. code-block:: python
+
+  from seam import Seam
+
+  seam = Seam()
+
+  paginator = seam.create_paginator(seam.connected_accounts.list, {"limit": 20})
+
+  for account in paginator.flatten():
+      print(account.account_type_display_name)
+
+Return all resources across all pages as a list
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+.. code-block:: python
+
+  from seam import Seam
+
+  seam = Seam()
+
+  paginator = seam.create_paginator(seam.connected_accounts.list, {"limit": 20})
+
+  all_connected_accounts = paginator.flatten_to_list()
+
 Interacting with Multiple Workspaces
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

seam/pagination.py

@@ -0,0 +1,10 @@
+class Pagination:
+    def __init__(
+        self,
+        has_next_page: bool,
+        next_page_cursor: str | None,
+        next_page_url: str | None,
+    ):
+        self.has_next_page = has_next_page
+        self.next_page_cursor = next_page_cursor
+        self.next_page_url = next_page_url

seam/paginator.py

@@ -0,0 +1,108 @@
+from typing import Callable, Dict, Any, Tuple, Generator, List
+from .client import SeamHttpClient
+from niquests import Response, JSONDecodeError
+from .pagination import Pagination
+
+
+class SeamPaginator:
+    """
+    Handles pagination for API list endpoints.
+
+    Iterates through pages of results returned by a callable function.
+    """
+
+    _FIRST_PAGE = "FIRST_PAGE"
+
+    def __init__(
+        self,
+        client: SeamHttpClient,
+        request: Callable,
+        params: Dict[str, Any] = None,
+    ):
+        """
+        Initializes the Paginator.
+
+        Args:
+            request: The function to call to fetch a page of data.
+            http_client: The Seam HTTP client used in the request.
+            params: Initial parameters to pass to the callable function.
+        """
+        self._request = request
+        self.client = client
+        self._params = params or {}
+        self._pagination_cache: Dict[str, Pagination] = {}
+
+    def first_page(self) -> Tuple[List[Any], Pagination | None]:
+        """Fetches the first page of results."""
+        self.client.hooks["response"].append(
+            lambda response: self._cache_pagination(response, self._FIRST_PAGE)
+        )
+        data = self._request(**self._params)
+        self.client.hooks["response"].pop()
+
+        pagination = self._pagination_cache.get(self._FIRST_PAGE)
+
+        return data, pagination
+
+    def next_page(
+        self, next_page_cursor: str, /
+    ) -> Tuple[List[Any], Pagination | None]:
+        """Fetches the next page of results using a cursor."""
+        if not next_page_cursor:
+            raise ValueError("Cannot get the next page with a null next_page_cursor.")
+
+        params = {
+            **self._params,
+            "page_cursor": next_page_cursor,
+        }
+
+        self.client.hooks["response"].append(
+            lambda response: self._cache_pagination(response, next_page_cursor)
+        )
+        data = self._request(**params)
+        self.client.hooks["response"].pop()
+
+        pagination = self._pagination_cache.get(next_page_cursor)
+
+        return data, pagination
+
+    def flatten_to_list(self) -> List[Any]:
+        """Fetches all pages and returns all items as a single list."""
+        all_items = []
+        current_items, pagination = self.first_page()
+
+        if current_items:
+            all_items.extend(current_items)
+
+        while pagination.has_next_page:
+            current_items, pagination = self.next_page(pagination.next_page_cursor)
+            if current_items:
+                all_items.extend(current_items)
+
+        return all_items
+
+    def flatten(self) -> Generator[Any, None, None]:
+        """Fetches all pages and yields items one by one using a generator."""
+        current_items, pagination = self.first_page()
+        if current_items:
+            yield from current_items
+
+        while pagination and pagination.has_next_page and pagination.next_page_cursor:
+            current_items, pagination = self.next_page(pagination.next_page_cursor)
+            if current_items:
+                yield from current_items
+
+    def _cache_pagination(self, response: Response, page_key: str) -> None:
+        """Extracts pagination dict from response, creates Pagination object, and caches it."""
+        try:
+            response_json = response.json()
+            pagination = response_json.get("pagination", {})
+        except JSONDecodeError:
+            pagination = {}
+
+        if isinstance(pagination, dict):
+            self._pagination_cache[page_key] = Pagination(
+                has_next_page=pagination.get("has_next_page", False),
+                next_page_cursor=pagination.get("next_page_cursor"),
+                next_page_url=pagination.get("next_page_url"),
+            )

seam/seam.py

@@ -1,4 +1,4 @@
-from typing import Any, Optional, Union, Dict
+from typing import Any, Optional, Union, Dict, Callable
 from typing_extensions import Self
 from urllib3.util.retry import Retry
 
@@ -7,6 +7,7 @@
 from .routes import Routes
 from .models import AbstractSeam
 from .client import SeamHttpClient
+from .paginator import SeamPaginator
 
 
 class Seam(AbstractSeam):
@@ -89,6 +90,31 @@ def __init__(
 
         Routes.__init__(self, client=self.client, defaults=self.defaults)
 
+    def create_paginator(
+        self, request: Callable, params: Optional[Dict[str, Any]] = None, /
+    ) -> SeamPaginator:
+        """
+        Creates a Paginator instance for iterating through list endpoints.
+
+        This is a helper method to simplify the process of paginating through
+        API results.
+
+        Args:
+            request: The API route method function to call for fetching pages
+                (e.g., connected_accounts.list).
+            params: Optional dictionary of initial parameters to pass to the request
+                function.
+
+        Returns:
+            An initialized Paginator object ready to fetch pages.
+
+        Example:
+            >>> connected_accounts_paginator = seam.create_paginator(seam.connected_accounts.list)
+            >>> for connected_account in connected_accounts_paginator.flatten():
+            >>>     print(connected_account.account_type_display_name)
+        """
+        return SeamPaginator(self.client, request, params)
+
     @classmethod
     def from_api_key(
         cls,

test/paginator_test.py

@@ -0,0 +1,49 @@
+from seam import Seam
+
+
+def test_paginator_first_page(seam: Seam):
+    paginator = seam.create_paginator(seam.connected_accounts.list, {"limit": 2})
+    connected_accounts, pagination = paginator.first_page()
+
+    assert isinstance(connected_accounts, list)
+    assert len(connected_accounts) == 2
+    assert pagination is not None
+    assert pagination.has_next_page is True
+    assert pagination.next_page_cursor is not None
+    assert pagination.next_page_url is not None
+
+
+def test_paginator_next_page(seam: Seam):
+    paginator = seam.create_paginator(seam.connected_accounts.list, {"limit": 2})
+    first_page_accounts, first_pagination = paginator.first_page()
+
+    assert len(first_page_accounts) == 2
+    assert first_pagination.has_next_page is True
+
+    next_page_accounts, _ = paginator.next_page(first_pagination.next_page_cursor)
+
+    assert isinstance(next_page_accounts, list)
+    assert len(next_page_accounts) == 1
+
+
+def test_paginator_flatten_to_list(seam: Seam):
+    all_connected_accounts = seam.connected_accounts.list()
+
+    paginator = seam.create_paginator(seam.connected_accounts.list, {"limit": 1})
+    paginated_accounts = paginator.flatten_to_list()
+
+    assert len(paginated_accounts) > 1
+    assert len(paginated_accounts) == len(all_connected_accounts)
+
+
+def test_paginator_flatten(seam: Seam):
+    all_connected_accounts = seam.connected_accounts.list()
+
+    paginator = seam.create_paginator(seam.connected_accounts.list, {"limit": 1})
+
+    collected_accounts = []
+    for account in paginator.flatten():
+        collected_accounts.append(account)
+
+    assert len(collected_accounts) > 1
+    assert len(collected_accounts) == len(all_connected_accounts)