Test paginator

Author: andrii-balitskyi

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[..., Any], 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, params)
+
     @classmethod
     def from_api_key(
         cls,

test/paginator_test.py

@@ -0,0 +1,60 @@
+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, params={"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, params={"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, params={"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)