feat: add PaginationHelper utility class

Author: LifeJiggy

src/gradient/_utils/__init__.py

@@ -36,6 +36,7 @@
     RateLimiter as RateLimiter,
     BatchProcessor as BatchProcessor,
     DataExporter as DataExporter,
+    PaginationHelper as PaginationHelper,
 )
 from ._compat import (
     get_args as get_args,

src/gradient/_utils/_utils.py

@@ -715,6 +715,129 @@ def export_csv(self, data: Any, file_path: str | None = None, headers: list[str]
                 output.close()
 
 
+# Pagination Classes
+class PaginationHelper:
+    """Helper for handling paginated API responses."""
+
+    def __init__(self, page_size: int = 20, max_pages: int | None = None) -> None:
+        """Initialize pagination helper.
+
+        Args:
+            page_size: Number of items per page
+            max_pages: Maximum number of pages to fetch (None for unlimited)
+        """
+        self.page_size: int = page_size
+        self.max_pages: int | None = max_pages
+
+    def paginate(self, fetch_func: Callable[[dict[str, Any]], Any], **kwargs: Any) -> list[Any]:
+        """Paginate through all results using the provided fetch function.
+
+        Args:
+            fetch_func: Function that takes pagination params and returns response
+            **kwargs: Additional parameters to pass to fetch_func
+
+        Returns:
+            List of all items across all pages
+        """
+        all_items = []
+        page = 1
+
+        while self.max_pages is None or page <= self.max_pages:
+            # Add pagination parameters
+            params = kwargs.copy()
+            params.update({
+                "page": page,
+                "per_page": self.page_size
+            })
+
+            try:
+                response = fetch_func(params)
+                items = self._extract_items(response)
+
+                if not items:
+                    break  # No more items
+
+                all_items.extend(items)
+
+                # Check if we got fewer items than requested (last page)
+                if len(items) < self.page_size:
+                    break
+
+                page += 1
+
+            except Exception as e:
+                # If it's a pagination error or no more pages, stop
+                if self._is_pagination_end_error(e):
+                    break
+                raise
+
+        return all_items
+
+    def _extract_items(self, response: Any) -> list[Any]:
+        """Extract items from API response."""
+        # Handle different response formats
+        if hasattr(response, 'data') and isinstance(response.data, list):
+            return response.data
+        elif hasattr(response, 'items') and isinstance(response.items, list):
+            return response.items
+        elif hasattr(response, 'results') and isinstance(response.results, list):
+            return response.results
+        elif isinstance(response, list):
+            return response
+        elif isinstance(response, dict):
+            # Try common keys
+            for key in ['data', 'items', 'results', 'objects']:
+                if key in response and isinstance(response[key], list):
+                    return response[key]
+        return []
+
+    def _is_pagination_end_error(self, error: Exception) -> bool:
+        """Check if error indicates end of pagination."""
+        error_str = str(error).lower()
+        return any(phrase in error_str for phrase in [
+            'page not found',
+            'invalid page',
+            'no more pages',
+            'pagination end'
+        ])
+
+    async def paginate_async(self, fetch_func: Callable[[dict[str, Any]], Any], **kwargs: Any) -> list[Any]:
+        """Async version of paginate."""
+        import asyncio
+
+        all_items = []
+        page = 1
+
+        while self.max_pages is None or page <= self.max_pages:
+            # Add pagination parameters
+            params = kwargs.copy()
+            params.update({
+                "page": page,
+                "per_page": self.page_size
+            })
+
+            try:
+                response = await fetch_func(params)
+                items = self._extract_items(response)
+
+                if not items:
+                    break
+
+                all_items.extend(items)
+
+                if len(items) < self.page_size:
+                    break
+
+                page += 1
+
+            except Exception as e:
+                if self._is_pagination_end_error(e):
+                    break
+                raise
+
+        return all_items
+
+
 # API Key Validation Functions
 def validate_api_key(api_key: str | None) -> bool:
     """Validate an API key format.

tests/test_pagination_helper.py

@@ -0,0 +1,110 @@
+"""Tests for pagination helper functionality."""
+
+import pytest
+from gradient._utils import PaginationHelper
+
+
+class TestPaginationHelper:
+    """Test pagination helper functionality."""
+
+    def test_pagination_helper_basic(self):
+        """Test basic pagination functionality."""
+        helper = PaginationHelper(page_size=2, max_pages=3)
+
+        # Mock responses with different pages
+        responses = [
+            {"data": ["item1", "item2"]},  # Page 1
+            {"data": ["item3", "item4"]},  # Page 2
+            {"data": ["item5"]},          # Page 3 (partial)
+        ]
+        response_index = 0
+
+        def mock_fetch(params):
+            nonlocal response_index
+            if response_index < len(responses):
+                result = responses[response_index]
+                response_index += 1
+                return result
+            return {"data": []}
+
+        result = helper.paginate(mock_fetch)
+        assert result == ["item1", "item2", "item3", "item4", "item5"]
+
+    def test_pagination_helper_max_pages(self):
+        """Test pagination with max pages limit."""
+        helper = PaginationHelper(page_size=2, max_pages=2)
+
+        responses = [
+            {"data": ["item1", "item2"]},
+            {"data": ["item3", "item4"]},
+            {"data": ["item5", "item6"]},  # Should not be reached
+        ]
+        response_index = 0
+
+        def mock_fetch(params):
+            nonlocal response_index
+            if response_index < len(responses):
+                result = responses[response_index]
+                response_index += 1
+                return result
+            return {"data": []}
+
+        result = helper.paginate(mock_fetch)
+        assert result == ["item1", "item2", "item3", "item4"]
+
+    def test_pagination_helper_empty_response(self):
+        """Test pagination with empty response."""
+        helper = PaginationHelper(page_size=2)
+
+        def mock_fetch(params):
+            return {"data": []}
+
+        result = helper.paginate(mock_fetch)
+        assert result == []
+
+    def test_pagination_helper_different_response_formats(self):
+        """Test pagination with different response formats."""
+        helper = PaginationHelper(page_size=2)
+
+        # Test different response formats
+        test_cases = [
+            {"data": ["item1", "item2"]},
+            {"items": ["item3", "item4"]},
+            {"results": ["item5", "item6"]},
+            ["item7", "item8"],  # Direct list
+        ]
+
+        for i, expected_format in enumerate(test_cases):
+            response_index = 0
+
+            def mock_fetch(params):
+                nonlocal response_index
+                if response_index == 0:
+                    response_index += 1
+                    return expected_format
+                return {"data": []}
+
+            result = helper.paginate(mock_fetch)
+            expected_items = expected_format if isinstance(expected_format, list) else expected_format.get(
+                list(expected_format.keys())[0], []
+            )
+            assert result == expected_items
+
+    def test_pagination_helper_extract_items(self):
+        """Test item extraction from different response formats."""
+        helper = PaginationHelper()
+
+        # Test various response formats
+        test_responses = [
+            type('MockResponse', (), {'data': ['item1']})(),
+            type('MockResponse', (), {'items': ['item2']})(),
+            type('MockResponse', (), {'results': ['item3']})(),
+            ['item4'],
+            {'data': ['item5']},
+            {'unknown': ['should_be_empty']},
+        ]
+
+        for i, response in enumerate(test_responses):
+            items = helper._extract_items(response)
+            expected_items = [f'item{i+1}'] if i < 5 else []
+            assert items == expected_items