diff --git a/README.rst b/README.rst index 17439871..727c4f4d 100644 --- a/README.rst +++ b/README.rst @@ -47,6 +47,16 @@ Contents * `Action Attempts `_ + * `Pagination `_ + + * `Manually fetch pages with the nextPageCursor `_ + + * `Resume pagination `_ + + * `Iterate over all resources `_ + + * `Return all resources across all pages as a list `_ + * `Interacting with Multiple Workspaces `_ * `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 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/seam/pagination.py b/seam/pagination.py new file mode 100644 index 00000000..52df67f1 --- /dev/null +++ b/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 diff --git a/seam/paginator.py b/seam/paginator.py new file mode 100644 index 00000000..fedb86d9 --- /dev/null +++ b/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"), + ) diff --git a/seam/seam.py b/seam/seam.py index be1c7bea..df49551e 100644 --- a/seam/seam.py +++ b/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, diff --git a/test/paginator_test.py b/test/paginator_test.py new file mode 100644 index 00000000..f8ba6fa9 --- /dev/null +++ b/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)