refactor: simplify pagination logic

Author: matsk-sinch

sinch/core/pagination.py

@@ -1,5 +1,4 @@
 from abc import ABC, abstractmethod
-from collections import namedtuple
 from typing import Generic
 from sinch.core.types import BM
 
@@ -50,16 +49,6 @@ async def __anext__(self):
 class Paginator(ABC, Generic[BM]):
     """
     Pagination response object.
-
-    auto_paging_iter method returns an iterator object that can be used for iterator-based page traversing.
-    For example:
-        for page in paginated_response.auto_paging_iter():
-            ...process page object
-
-    For manual pagination use has_next_page property with next_page() method.
-    For example:
-        if paginated_response.has_next_page:
-            paginated_response = paginated_response.next_page()
     """
     def __init__(self, sinch, endpoint, result: BM):
         self._sinch = sinch
@@ -81,11 +70,6 @@ def content(self):
     def iterator(self):
         pass
 
-    # TODO: Make get_content() method abstract in Parent class as we implement in the other domains:
-    #  - Refactor pydantic models in other domains to have a content property.
-    def get_content(self):
-        pass
-
     @abstractmethod
     def next_page(self):
         pass
@@ -145,7 +129,7 @@ async def _initialize(cls, sinch, endpoint):
 class TokenBasedPaginator(Paginator[BM]):
     """Base paginator for token-based pagination with explicit page navigation and metadata."""
 
-    def __init__(self, sinch, endpoint, yield_first_page=False, result=None):
+    def __init__(self, sinch, endpoint, result=None):
         super().__init__(sinch, endpoint, result or sinch.configuration.transport.request(endpoint))
 
     def content(self) -> list[BM]:
@@ -157,9 +141,9 @@ def next_page(self):
             return None
 
         self.endpoint.request_data.page_token = self.result.next_page_token
-        next_result = self._sinch.configuration.transport.request(self.endpoint)
-
-        return self.__class__(self._sinch, self.endpoint, result=next_result)
+        self.result = self._sinch.configuration.transport.request(self.endpoint)
+        self._calculate_next_page()
+        return self
 
     def iterator(self):
         """Iterates over individual items across all pages."""
@@ -172,52 +156,14 @@ def iterator(self):
                 break
             paginator = next_page_instance
 
-    def get_content(self):
-        """Returns structured pagination metadata along with the first page's content (sync)."""
-        next_page_instance = self.next_page()
-        return self._get_content(next_page_instance, sync=True)
-
-    def _get_content(self, next_page_instance, sync=True):
-        """Core logic for `get_content()`, shared between sync and async versions."""
-        PagedListResponse = namedtuple(
-            "PagedResponse", ["result", "has_next_page", "next_page_info", "next_page"]
-        )
-
-        next_page_info = {
-            "result": self.content(),
-            "result.next": (
-                self.content() + (next_page_instance.content() if next_page_instance else [])
-            ),
-            "has_next_page": self.has_next_page,
-            "has_next_page.next": bool(next_page_instance and next_page_instance.has_next_page),
-        }
-
-        next_page_wrapper = self._get_next_page_wrapper(next_page_instance, sync)
-
-        return PagedListResponse(
-            result=self.content(),
-            has_next_page=self.has_next_page,
-            next_page_info=next_page_info,
-            next_page=next_page_wrapper
-        )
-
-    def _get_next_page_wrapper(self, next_page_instance, sync):
-        """Returns a function for fetching the next page."""
-        if sync:
-            return lambda: next_page_instance.get_content() if next_page_instance else None
-        else:
-            async def async_next_page_wrapper():
-                return await next_page_instance.get_content() if next_page_instance else None
-            return async_next_page_wrapper
-
     def _calculate_next_page(self):
         self.has_next_page = bool(getattr(self.result, "next_page_token", None))
 
     @classmethod
     def _initialize(cls, sinch, endpoint):
         """Creates an instance of the paginator skipping first page."""
         result = sinch.configuration.transport.request(endpoint)
-        return cls(sinch, endpoint, yield_first_page=False, result=result)
+        return cls(sinch, endpoint, result=result)
 
 
 class AsyncTokenBasedPaginator(TokenBasedPaginator):
@@ -244,12 +190,7 @@ async def iterator(self):
                 break
             paginator = next_page_instance
 
-    async def get_content(self):
-        """Returns structured pagination metadata"""
-        next_page_instance = await self.next_page()
-        return self._get_content(next_page_instance, sync=False)
-
     @classmethod
     async def _initialize(cls, sinch, endpoint):
         result = await sinch.configuration.transport.request(endpoint)
-        return cls(sinch, endpoint, yield_first_page=False, result=result)
+        return cls(sinch, endpoint, result=result)

sinch/domains/numbers/active_numbers.py

@@ -136,7 +136,7 @@ class ActiveNumbersWithAsyncPagination(ActiveNumbers):
     async def list(
         self,
         region_code: StrictStr,
-        number_type: StrictStr,
+        number_type: NumberTypeValues,
         number_pattern: Optional[StrictStr] = None,
         number_search_pattern: Optional[NumberSearchPatternTypeValues] = None,
         capabilities: Optional[CapabilityTypeValuesList] = None,

tests/conftest.py

@@ -186,30 +186,6 @@ def token_based_pagination_request_data():
     )
 
 
-@pytest.fixture
-def first_token_based_pagination_response():
-    return TokenBasedPaginationResponse(
-        pig_dogs=["Walaszek", "Połać"],
-        next_page_token="za30%wsze"
-    )
-
-
-@pytest.fixture
-def second_token_based_pagination_response():
-    return TokenBasedPaginationResponse(
-        pig_dogs=["Bartosz", "Piotr"],
-        next_page_token="ka#556"
-    )
-
-
-@pytest.fixture
-def third_token_based_pagination_response():
-    return TokenBasedPaginationResponse(
-        pig_dogs=["Madrid", "Spain"],
-        next_page_token=""
-    )
-
-
 @pytest.fixture
 def int_based_pagination_request_data():
     return IntBasedPaginationRequest(
@@ -334,4 +310,10 @@ def mock_pagination_active_number_responses():
              next_page_token="token_2"),
         Mock(content=[ActiveNumber(phone_number="+12345678905")],
              next_page_token=None)
+    ]
+
+@pytest.fixture
+def mock_pagination_expected_phone_numbers_response():
+    return [
+        "+12345678901", "+12345678902", "+12345678903", "+12345678904", "+12345678905"
     ]

tests/e2e/numbers/features/steps/numbers.steps.py

@@ -165,13 +165,13 @@ def step_when_list_phone_numbers(context):
         number_type='LOCAL'
     )
     # Get the first page
-    context.response = response.get_content()
+    context.response = response.content()
 
 
 @then('the response contains "{count}" phone numbers')
 def step_then_response_contains_x_phone_numbers(context, count):
-    assert len(context.response.result) == int(count), \
-        f'Expected {count}, got {len(context.response.data)}'
+    assert len(context.response) == int(count), \
+        f'Expected {count}, got {len(context.response)}'
 
 @when("I send a request to list all the phone numbers")
 def step_when_list_all_phone_numbers(context):

tests/unit/domains/numbers/models/active/requests/test_list_active_numbers_request_model.py

@@ -43,7 +43,7 @@ def test_list_active_numbers_request_expects_parsed_input():
         "number_search_pattern": "START",
         "number_pattern": "5678",
         "page_token": "abc123",
-        "order_by": "phoneNumber"
+        "order_by": "PHONE_NUMBER"
     }
 
     request = ListActiveNumbersRequest(**data)