feat: Add seam.create_paginator

seam/paginator.py

@@ -0,0 +1,117 @@
+from typing import Callable, Dict, Any, Tuple, Generator, List
+from .client import SeamHttpClient
+from niquests import Response, JSONDecodeError
+
+
+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
+
+
+class Paginator:
+    """
+    Handles pagination for API list endpoints.
+
+    Iterates through pages of results returned by a callable function.
+    """
+
+    _FIRST_PAGE = "FIRST_PAGE"
+
+    def __init__(
+        self,
+        request: Callable,
+        http_client: SeamHttpClient,
+        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._http_client = http_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._http_client.hooks["response"].append(
+            lambda response: self._cache_pagination(response, self._FIRST_PAGE)
+        )
+        data = self._request(**self._params)
+        self._http_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._http_client.hooks["response"].append(
+            lambda response: self._cache_pagination(response, next_page_cursor)
+        )
+        data = self._request(**params)
+        self._http_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 Paginator
 
 
 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
+    ) -> Paginator:
+        """
+        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 Paginator(request, self.client, 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)