feat: Token pagination support #896
Open
+839
−0
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…ython into token-pagination-support
|
14 tasks
Contributor
There was a problem hiding this comment.
Pull request overview
This pull request adds token-based pagination support to the Twilio Python SDK as an alternative to the existing page-number-based pagination. The new TokenPagination class extends the base Page class and uses nextToken and previousToken values from API responses to navigate between pages, which is a common pattern for APIs that need efficient pagination over large datasets.
Key Changes:
- Introduced
TokenPaginationclass that inherits fromPageand implements token-based navigation - Added comprehensive unit tests covering properties, navigation, streaming, error handling, and async operations
- Implemented both synchronous and asynchronous page navigation methods
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 5 comments.
| File | Description |
|---|---|
twilio/base/token_pagination.py |
New pagination class implementing token-based pagination with properties for key, page_size, next_token, and previous_token; includes sync and async navigation methods |
tests/unit/base/test_token_pagination.py |
Comprehensive test suite with 600+ lines covering property accessors, page navigation, streaming, error cases, and async operations |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+1
to
+691
| import unittest | ||
| from unittest.mock import Mock, AsyncMock | ||
| from tests import IntegrationTestCase | ||
| from tests.holodeck import Request | ||
| from twilio.base.exceptions import TwilioException | ||
| from twilio.base.token_pagination import TokenPagination | ||
| from twilio.http.response import Response | ||
|
|
||
|
|
||
| class MockTokenPaginationPage(TokenPagination): | ||
| """Mock implementation of TokenPagination for testing""" | ||
|
|
||
| def get_instance(self, payload): | ||
| return payload | ||
|
|
||
|
|
||
| class TokenPaginationPropertyTest(unittest.TestCase): | ||
| """Test TokenPagination property accessors""" | ||
|
|
||
| def setUp(self): | ||
| self.version = Mock() | ||
| self.version.domain = Mock() | ||
|
|
||
| # Mock response with token pagination format | ||
| self.payload = { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 50, | ||
| "nextToken": "next_abc123", | ||
| "previousToken": "prev_xyz789", | ||
| }, | ||
| "items": [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}], | ||
| } | ||
|
|
||
| self.response = Mock() | ||
| self.response.text = """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 50, | ||
| "nextToken": "next_abc123", | ||
| "previousToken": "prev_xyz789" | ||
| }, | ||
| "items": [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}] | ||
| } | ||
| """ | ||
| self.response.status_code = 200 | ||
|
|
||
| self.solution = { | ||
| "account_sid": "ACxxxx", | ||
| "api_uri": "/Accounts/ACxxxx/Resources.json", | ||
| } | ||
| self.page = MockTokenPaginationPage( | ||
| self.version, | ||
| self.response, | ||
| "/Accounts/ACxxxx/Resources.json", | ||
| self.solution, | ||
| ) | ||
|
|
||
| def test_key_property(self): | ||
| """Test that key property returns the correct value""" | ||
| self.assertEqual(self.page.key, "items") | ||
|
|
||
| def test_page_size_property(self): | ||
| """Test that page_size property returns the correct value""" | ||
| self.assertEqual(self.page.page_size, 50) | ||
|
|
||
| def test_next_token_property(self): | ||
| """Test that next_token property returns the correct value""" | ||
| self.assertEqual(self.page.next_token, "next_abc123") | ||
|
|
||
| def test_previous_token_property(self): | ||
| """Test that previous_token property returns the correct value""" | ||
| self.assertEqual(self.page.previous_token, "prev_xyz789") | ||
|
|
||
| def test_properties_without_meta(self): | ||
| """Test that properties return None when meta is missing""" | ||
| response = Mock() | ||
| response.text = '{"items": []}' | ||
| response.status_code = 200 | ||
|
|
||
| page = MockTokenPaginationPage( | ||
| self.version, response, "/Accounts/ACxxxx/Resources.json", self.solution | ||
| ) | ||
|
|
||
| self.assertIsNone(page.key) | ||
| self.assertIsNone(page.page_size) | ||
| self.assertIsNone(page.next_token) | ||
| self.assertIsNone(page.previous_token) | ||
|
|
||
| def test_properties_with_partial_meta(self): | ||
| """Test that properties return None when specific keys are missing""" | ||
| response = Mock() | ||
| response.text = '{"meta": {"key": "items"}, "items": []}' | ||
| response.status_code = 200 | ||
|
|
||
| page = MockTokenPaginationPage( | ||
| self.version, response, "/Accounts/ACxxxx/Resources.json", self.solution | ||
| ) | ||
|
|
||
| self.assertEqual(page.key, "items") | ||
| self.assertIsNone(page.page_size) | ||
| self.assertIsNone(page.next_token) | ||
| self.assertIsNone(page.previous_token) | ||
|
|
||
|
|
||
| class TokenPaginationNavigationTest(IntegrationTestCase): | ||
| """Test TokenPagination next_page and previous_page methods""" | ||
|
|
||
| def setUp(self): | ||
| super(TokenPaginationNavigationTest, self).setUp() | ||
|
|
||
| # Mock first page response | ||
| self.holodeck.mock( | ||
| Response( | ||
| 200, | ||
| """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 2, | ||
| "nextToken": "token_page2", | ||
| "previousToken": null | ||
| }, | ||
| "items": [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}] | ||
| } | ||
| """, | ||
| ), | ||
| Request( | ||
| url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Resources.json" | ||
| ), | ||
| ) | ||
|
|
||
| # Mock second page response (next page) | ||
| self.holodeck.mock( | ||
| Response( | ||
| 200, | ||
| """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 2, | ||
| "nextToken": "token_page3", | ||
| "previousToken": "token_prev1" | ||
| }, | ||
| "items": [{"id": 3, "name": "Item 3"}, {"id": 4, "name": "Item 4"}] | ||
| } | ||
| """, | ||
| ), | ||
| Request( | ||
| url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Resources.json?pageToken=token_page2" | ||
| ), | ||
| ) | ||
|
|
||
| # Mock third page response (has no next) | ||
| self.holodeck.mock( | ||
| Response( | ||
| 200, | ||
| """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 2, | ||
| "nextToken": null, | ||
| "previousToken": "token_prev2" | ||
| }, | ||
| "items": [{"id": 5, "name": "Item 5"}] | ||
| } | ||
| """, | ||
| ), | ||
| Request( | ||
| url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Resources.json?pageToken=token_page3" | ||
| ), | ||
| ) | ||
|
|
||
| # Mock previous page response (going back to page 1) | ||
| self.holodeck.mock( | ||
| Response( | ||
| 200, | ||
| """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 2, | ||
| "nextToken": "token_page2", | ||
| "previousToken": null | ||
| }, | ||
| "items": [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}] | ||
| } | ||
| """, | ||
| ), | ||
| Request( | ||
| url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Resources.json?pageToken=token_prev1" | ||
| ), | ||
| ) | ||
|
|
||
| # Mock going back from page 3 to page 2 | ||
| self.holodeck.mock( | ||
| Response( | ||
| 200, | ||
| """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 2, | ||
| "nextToken": "token_page3", | ||
| "previousToken": "token_prev1" | ||
| }, | ||
| "items": [{"id": 3, "name": "Item 3"}, {"id": 4, "name": "Item 4"}] | ||
| } | ||
| """, | ||
| ), | ||
| Request( | ||
| url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Resources.json?pageToken=token_prev2" | ||
| ), | ||
| ) | ||
|
|
||
| self.version = self.client.api.v2010 | ||
| self.response = self.version.page( | ||
| method="GET", uri="/Accounts/ACaaaa/Resources.json" | ||
| ) | ||
|
|
||
| self.solution = { | ||
| "account_sid": "ACaaaa", | ||
| "api_uri": "/2010-04-01/Accounts/ACaaaa/Resources.json", | ||
| } | ||
|
|
||
| self.page = MockTokenPaginationPage( | ||
| self.version, | ||
| self.response, | ||
| "/2010-04-01/Accounts/ACaaaa/Resources.json", | ||
| self.solution, | ||
| ) | ||
|
|
||
| def test_next_page(self): | ||
| """Test that next_page() navigates to the next page using token""" | ||
| self.assertIsNotNone(self.page.next_token) | ||
| self.assertEqual(self.page.next_token, "token_page2") | ||
|
|
||
| next_page = self.page.next_page() | ||
|
|
||
| self.assertIsNotNone(next_page) | ||
| self.assertIsInstance(next_page, MockTokenPaginationPage) | ||
| # Verify we got the next page's data | ||
| self.assertEqual(next_page.next_token, "token_page3") | ||
| self.assertEqual(next_page.previous_token, "token_prev1") | ||
|
|
||
| def test_next_page_none_when_no_token(self): | ||
| """Test that next_page() returns None when there's no next token""" | ||
| # Navigate to the last page | ||
| next_page = self.page.next_page() | ||
| last_page = next_page.next_page() | ||
|
|
||
| # Last page should have no next token | ||
| self.assertIsNone(last_page.next_token) | ||
|
|
||
| # next_page() should return None | ||
| result = last_page.next_page() | ||
| self.assertIsNone(result) | ||
|
|
||
| def test_previous_page(self): | ||
| """Test that previous_page() navigates to the previous page using token""" | ||
| # Navigate to second page first | ||
| next_page = self.page.next_page() | ||
| self.assertIsNotNone(next_page.previous_token) | ||
| self.assertEqual(next_page.previous_token, "token_prev1") | ||
|
|
||
| # Go back to previous page | ||
| prev_page = next_page.previous_page() | ||
|
|
||
| self.assertIsNotNone(prev_page) | ||
| self.assertIsInstance(prev_page, MockTokenPaginationPage) | ||
| # Verify we got the first page's data | ||
| self.assertIsNone(prev_page.previous_token) | ||
| self.assertEqual(prev_page.next_token, "token_page2") | ||
|
|
||
| def test_previous_page_none_when_no_token(self): | ||
| """Test that previous_page() returns None when there's no previous token""" | ||
| # First page should have no previous token | ||
| self.assertIsNone(self.page.previous_token) | ||
|
|
||
| # previous_page() should return None | ||
| result = self.page.previous_page() | ||
| self.assertIsNone(result) | ||
|
|
||
| def test_navigation_chain(self): | ||
| """Test navigating through multiple pages forward and backward""" | ||
| # Page 1 -> Page 2 | ||
| page2 = self.page.next_page() | ||
| self.assertEqual(page2.previous_token, "token_prev1") | ||
|
|
||
| # Page 2 -> Page 3 | ||
| page3 = page2.next_page() | ||
| self.assertIsNone(page3.next_token) | ||
| self.assertEqual(page3.previous_token, "token_prev2") | ||
|
|
||
| # Page 3 -> Page 2 (backward) | ||
| back_to_page2 = page3.previous_page() | ||
| self.assertIsNotNone(back_to_page2) | ||
|
|
||
|
|
||
| class TokenPaginationErrorTest(unittest.TestCase): | ||
| """Test TokenPagination error handling""" | ||
|
|
||
| def test_next_page_without_uri_in_solution(self): | ||
| """Test that next_page() raises error when URI is missing""" | ||
| version = Mock() | ||
| response = Mock() | ||
| response.text = """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 50, | ||
| "nextToken": "abc123" | ||
| }, | ||
| "items": [] | ||
| } | ||
| """ | ||
| response.status_code = 200 | ||
|
|
||
| # Solution without URI | ||
| solution = {"account_sid": "ACxxxx"} | ||
|
|
||
| # Pass empty string as URI to test the error case | ||
| page = MockTokenPaginationPage(version, response, "", solution) | ||
|
|
||
| with self.assertRaises(TwilioException) as context: | ||
| page.next_page() | ||
|
|
||
| self.assertIn("URI must be provided", str(context.exception)) | ||
|
|
||
|
|
||
| class TokenPaginationStreamTest(IntegrationTestCase): | ||
| """Test streaming with TokenPagination""" | ||
|
|
||
| def setUp(self): | ||
| super(TokenPaginationStreamTest, self).setUp() | ||
|
|
||
| # Mock page 1 | ||
| self.holodeck.mock( | ||
| Response( | ||
| 200, | ||
| """ | ||
| { | ||
| "meta": { | ||
| "key": "records", | ||
| "pageSize": 2, | ||
| "nextToken": "token_2" | ||
| }, | ||
| "records": [{"id": 1}, {"id": 2}] | ||
| } | ||
| """, | ||
| ), | ||
| Request( | ||
| url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Records.json" | ||
| ), | ||
| ) | ||
|
|
||
| # Mock page 2 | ||
| self.holodeck.mock( | ||
| Response( | ||
| 200, | ||
| """ | ||
| { | ||
| "meta": { | ||
| "key": "records", | ||
| "pageSize": 2, | ||
| "nextToken": "token_3", | ||
| "previousToken": "token_2" | ||
| }, | ||
| "records": [{"id": 3}, {"id": 4}] | ||
| } | ||
| """, | ||
| ), | ||
| Request( | ||
| url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Records.json?pageToken=token_2" | ||
| ), | ||
| ) | ||
|
|
||
| # Mock page 3 (final) | ||
| self.holodeck.mock( | ||
| Response( | ||
| 200, | ||
| """ | ||
| { | ||
| "meta": { | ||
| "key": "records", | ||
| "pageSize": 2, | ||
| "nextToken": null, | ||
| "previousToken": "token_3" | ||
| }, | ||
| "records": [{"id": 5}] | ||
| } | ||
| """, | ||
| ), | ||
| Request( | ||
| url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Records.json?pageToken=token_3" | ||
| ), | ||
| ) | ||
|
|
||
| self.version = self.client.api.v2010 | ||
| self.response = self.version.page( | ||
| method="GET", uri="/Accounts/ACaaaa/Records.json" | ||
| ) | ||
|
|
||
| self.solution = { | ||
| "account_sid": "ACaaaa", | ||
| "api_uri": "/2010-04-01/Accounts/ACaaaa/Records.json", | ||
| } | ||
|
|
||
| self.page = MockTokenPaginationPage( | ||
| self.version, | ||
| self.response, | ||
| "/2010-04-01/Accounts/ACaaaa/Records.json", | ||
| self.solution, | ||
| ) | ||
|
|
||
| def test_stream_all_records(self): | ||
| """Test streaming through all pages""" | ||
| records = list(self.version.stream(self.page)) | ||
|
|
||
| self.assertEqual(len(records), 5) | ||
| self.assertEqual(records[0]["id"], 1) | ||
| self.assertEqual(records[4]["id"], 5) | ||
|
|
||
| def test_stream_with_limit(self): | ||
| """Test streaming with a limit""" | ||
| records = list(self.version.stream(self.page, limit=3)) | ||
|
|
||
| self.assertEqual(len(records), 3) | ||
| self.assertEqual(records[0]["id"], 1) | ||
| self.assertEqual(records[2]["id"], 3) | ||
|
|
||
| def test_stream_with_page_limit(self): | ||
| """Test streaming with page limit""" | ||
| records = list(self.version.stream(self.page, page_limit=1)) | ||
|
|
||
| # Only first page (2 records) | ||
| self.assertEqual(len(records), 2) | ||
|
|
||
|
|
||
| class TokenPaginationInternalMethodTest(unittest.TestCase): | ||
| """Test TokenPagination internal methods""" | ||
|
|
||
| def setUp(self): | ||
| self.version = Mock() | ||
| self.version.domain = Mock() | ||
| self.version.domain.twilio = Mock() | ||
| self.version.domain.absolute_url = Mock( | ||
| side_effect=lambda uri: f"https://api.twilio.com{uri}" | ||
| ) | ||
|
|
||
| # Mock first page response | ||
| self.first_response = Mock() | ||
| self.first_response.text = """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 2, | ||
| "nextToken": "token_page2", | ||
| "previousToken": null | ||
| }, | ||
| "items": [{"id": 1}, {"id": 2}] | ||
| } | ||
| """ | ||
| self.first_response.status_code = 200 | ||
|
|
||
| # Mock next page response | ||
| self.next_response = Mock() | ||
| self.next_response.text = """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 2, | ||
| "nextToken": null, | ||
| "previousToken": "token_prev" | ||
| }, | ||
| "items": [{"id": 3}, {"id": 4}] | ||
| } | ||
| """ | ||
| self.next_response.status_code = 200 | ||
|
|
||
| self.solution = {"account_sid": "ACxxxx"} | ||
| self.page = MockTokenPaginationPage( | ||
| self.version, | ||
| self.first_response, | ||
| "/2010-04-01/Accounts/ACxxxx/Resources.json", | ||
| self.solution, | ||
| ) | ||
|
|
||
| def test_get_page_with_valid_token(self): | ||
| """Test _get_page() with a valid token""" | ||
| self.version.domain.twilio.request = Mock(return_value=self.next_response) | ||
|
|
||
| result = self.page._get_page("token_page2") | ||
|
|
||
| self.assertIsNotNone(result) | ||
| self.assertIsInstance(result, MockTokenPaginationPage) | ||
| self.version.domain.twilio.request.assert_called_once_with( | ||
| "GET", | ||
| "https://api.twilio.com/2010-04-01/Accounts/ACxxxx/Resources.json?pageToken=token_page2", | ||
| ) | ||
|
|
||
| def test_get_page_with_none_token(self): | ||
| """Test _get_page() with None token returns None""" | ||
| result = self.page._get_page(None) | ||
| self.assertIsNone(result) | ||
|
|
||
| def test_get_page_without_uri(self): | ||
| """Test _get_page() raises error when URI is missing""" | ||
| page = MockTokenPaginationPage( | ||
| self.version, self.first_response, "", self.solution | ||
| ) | ||
|
|
||
| with self.assertRaises(TwilioException) as context: | ||
| page._get_page("some_token") | ||
|
|
||
| self.assertIn("URI must be provided", str(context.exception)) | ||
|
|
||
| def test_repr(self): | ||
| """Test __repr__ method returns correct string""" | ||
| self.assertEqual(repr(self.page), "<TokenPagination>") | ||
|
|
||
|
|
||
| class TokenPaginationAsyncTest(unittest.IsolatedAsyncioTestCase): | ||
| """Test TokenPagination async methods""" | ||
|
|
||
| def setUp(self): | ||
| self.version = Mock() | ||
| self.version.domain = Mock() | ||
| self.version.domain.twilio = Mock() | ||
| self.version.domain.absolute_url = Mock( | ||
| side_effect=lambda uri: f"https://api.twilio.com{uri}" | ||
| ) | ||
|
|
||
| # Mock first page response | ||
| self.first_response = Mock() | ||
| self.first_response.text = """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 2, | ||
| "nextToken": "token_page2", | ||
| "previousToken": null | ||
| }, | ||
| "items": [{"id": 1}, {"id": 2}] | ||
| } | ||
| """ | ||
| self.first_response.status_code = 200 | ||
|
|
||
| # Mock next page response | ||
| self.next_response = Mock() | ||
| self.next_response.text = """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 2, | ||
| "nextToken": "token_page3", | ||
| "previousToken": "token_prev" | ||
| }, | ||
| "items": [{"id": 3}, {"id": 4}] | ||
| } | ||
| """ | ||
| self.next_response.status_code = 200 | ||
|
|
||
| # Mock previous page response | ||
| self.prev_response = Mock() | ||
| self.prev_response.text = """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 2, | ||
| "nextToken": "token_page2", | ||
| "previousToken": null | ||
| }, | ||
| "items": [{"id": 1}, {"id": 2}] | ||
| } | ||
| """ | ||
| self.prev_response.status_code = 200 | ||
|
|
||
| self.solution = {"account_sid": "ACxxxx"} | ||
|
|
||
| # Page with next token | ||
| self.page = MockTokenPaginationPage( | ||
| self.version, | ||
| self.first_response, | ||
| "/2010-04-01/Accounts/ACxxxx/Resources.json", | ||
| self.solution, | ||
| ) | ||
|
|
||
| # Page with previous token (page 2) | ||
| self.page_with_prev = MockTokenPaginationPage( | ||
| self.version, | ||
| self.next_response, | ||
| "/2010-04-01/Accounts/ACxxxx/Resources.json", | ||
| self.solution, | ||
| ) | ||
|
|
||
| async def test_get_page_async_with_valid_token(self): | ||
| """Test _get_page_async() with a valid token""" | ||
| self.version.domain.twilio.request_async = AsyncMock( | ||
| return_value=self.next_response | ||
| ) | ||
|
|
||
| result = await self.page._get_page_async("token_page2") | ||
|
|
||
| self.assertIsNotNone(result) | ||
| self.assertIsInstance(result, MockTokenPaginationPage) | ||
| self.version.domain.twilio.request_async.assert_called_once_with( | ||
| "GET", | ||
| "https://api.twilio.com/2010-04-01/Accounts/ACxxxx/Resources.json?pageToken=token_page2", | ||
| ) | ||
|
|
||
| async def test_get_page_async_with_none_token(self): | ||
| """Test _get_page_async() with None token returns None""" | ||
| result = await self.page._get_page_async(None) | ||
| self.assertIsNone(result) | ||
|
|
||
| async def test_get_page_async_without_uri(self): | ||
| """Test _get_page_async() raises error when URI is missing""" | ||
| page = MockTokenPaginationPage( | ||
| self.version, self.first_response, "", self.solution | ||
| ) | ||
|
|
||
| with self.assertRaises(TwilioException) as context: | ||
| await page._get_page_async("some_token") | ||
|
|
||
| self.assertIn("URI must be provided", str(context.exception)) | ||
|
|
||
| async def test_next_page_async(self): | ||
| """Test next_page_async() navigates to next page""" | ||
| self.version.domain.twilio.request_async = AsyncMock( | ||
| return_value=self.next_response | ||
| ) | ||
|
|
||
| next_page = await self.page.next_page_async() | ||
|
|
||
| self.assertIsNotNone(next_page) | ||
| self.assertIsInstance(next_page, MockTokenPaginationPage) | ||
| self.assertEqual(next_page.next_token, "token_page3") | ||
| self.assertEqual(next_page.previous_token, "token_prev") | ||
|
|
||
| async def test_next_page_async_none_when_no_token(self): | ||
| """Test next_page_async() returns None when there's no next token""" | ||
| # Create page with no next token | ||
| no_next_response = Mock() | ||
| no_next_response.text = """ | ||
| { | ||
| "meta": { | ||
| "key": "items", | ||
| "pageSize": 2, | ||
| "nextToken": null, | ||
| "previousToken": "token_prev" | ||
| }, | ||
| "items": [{"id": 5}] | ||
| } | ||
| """ | ||
| no_next_response.status_code = 200 | ||
|
|
||
| page = MockTokenPaginationPage( | ||
| self.version, | ||
| no_next_response, | ||
| "/2010-04-01/Accounts/ACxxxx/Resources.json", | ||
| self.solution, | ||
| ) | ||
|
|
||
| result = await page.next_page_async() | ||
| self.assertIsNone(result) | ||
|
|
||
| async def test_previous_page_async(self): | ||
| """Test previous_page_async() navigates to previous page""" | ||
| self.version.domain.twilio.request_async = AsyncMock( | ||
| return_value=self.prev_response | ||
| ) | ||
|
|
||
| prev_page = await self.page_with_prev.previous_page_async() | ||
|
|
||
| self.assertIsNotNone(prev_page) | ||
| self.assertIsInstance(prev_page, MockTokenPaginationPage) | ||
| self.assertIsNone(prev_page.previous_token) | ||
| self.assertEqual(prev_page.next_token, "token_page2") | ||
|
|
||
| async def test_previous_page_async_none_when_no_token(self): | ||
| """Test previous_page_async() returns None when there's no previous token""" | ||
| # First page has no previous token | ||
| result = await self.page.previous_page_async() | ||
| self.assertIsNone(result) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| unittest.main() |
There was a problem hiding this comment.
Consider adding test cases for edge cases:
- URIs that already contain query parameters (e.g.,
/path?existing=value) - Tokens with special characters that need URL encoding (e.g., tokens containing
+,=,/) - Empty tokens or tokens with only whitespace
These scenarios would help ensure the implementation handles real-world pagination edge cases correctly.
| raise TwilioException("URI must be provided for token pagination") | ||
|
|
||
| # Construct full URL with pageToken parameter | ||
| uri_with_token = f"{self._uri}?pageToken={token}" |
There was a problem hiding this comment.
The URL construction has two issues:
-
Missing URL encoding: The token value should be URL-encoded to handle special characters (e.g.,
+,=,/, etc.) that may appear in pagination tokens. -
Assumes no existing query parameters: If
self._urialready contains query parameters (e.g.,/path?existing=value), this will create an invalid URL like/path?existing=value?pageToken=token.
Recommended fix:
from urllib.parse import quote
# Check if URI already has query parameters
separator = '&' if '?' in self._uri else '?'
uri_with_token = f"{self._uri}{separator}pageToken={quote(token, safe='')}"| raise TwilioException("URI must be provided for token pagination") | ||
|
|
||
| # Construct full URL with pageToken parameter | ||
| uri_with_token = f"{self._uri}?pageToken={token}" |
There was a problem hiding this comment.
Same URL construction issues as line 83:
- Missing URL encoding: The token value should be URL-encoded.
- Assumes no existing query parameters: May create invalid URLs if the URI already has query parameters.
Apply the same fix:
from urllib.parse import quote
separator = '&' if '?' in self._uri else '?'
uri_with_token = f"{self._uri}{separator}pageToken={quote(token, safe='')}"
Comment on lines
+1
to
+4
| from typing import Optional | ||
|
|
||
| from twilio.base.exceptions import TwilioException | ||
| from twilio.base.page import Page |
There was a problem hiding this comment.
Missing import for URL encoding functionality. To properly handle URL construction with query parameters and special characters, add:
from urllib.parse import quoteThis import is needed to fix the URL construction issues in lines 83 and 105.
| } | ||
| """ | ||
|
|
||
| def __init__(self, version, response, uri: str, solution={}): |
There was a problem hiding this comment.
The response parameter lacks a type annotation. The parent class Page uses response: Response. Consider adding the same type hint for consistency:
from twilio.http.response import Response
def __init__(self, version, response: Response, uri: str, solution={}):
tiwarishubham635
requested changes
Dec 8, 2025
| raise TwilioException("URI must be provided for token pagination") | ||
|
|
||
| # Construct full URL with pageToken parameter | ||
| uri_with_token = f"{self._uri}?pageToken={token}" |
Contributor
There was a problem hiding this comment.
I see while creating this url we are not adding pagesize. Will this still work?
kridai
added a commit
to twilio/twilio-oai-generator
that referenced
this pull request
Dec 8, 2025
<!-- We appreciate the effort for this pull request but before that please make sure you read the contribution guidelines, then fill out the blanks below. Please format the PR title appropriately based on the type of change: <type>[!]: <description> Where <type> is one of: docs, chore, feat, fix, test, misc. Add a '!' after the type for breaking changes (e.g. feat!: new breaking feature). **All third-party contributors acknowledge that any contributions they provide will be made under the same open-source license that the open-source project is provided under.** Please enter each Issue number you are resolving in your PR after one of the following words [Fixes, Closes, Resolves]. This will auto-link these issues and close them when this PR is merged! e.g. Fixes #1 Closes #2 --> # Fixes # PR for python twilio/twilio-python#896 As part of this PR Added method setIsV1ApiStandard to TwilioCodegenAdapter, so that it is accessible across different generators Added inheriting TokenPagination class when IsV1Api is true ### Checklist - [x] I acknowledge that all my contributions will be made under the project's license - [ ] Run `make test-docker` - [ ] Verify affected language according to the code change: - [ ] Generate [twilio-java](https://github.com/twilio/twilio-java) from our [OpenAPI specification](https://github.com/twilio/twilio-oai) using the [scripts/build_twilio_library.py](./scripts/build_twilio_library.py) using `python scripts/build_twilio_library.py path/to/twilio-oai/spec/yaml path/to/twilio-java -l java` and inspect the diff - [ ] Run `make test` in `twilio-java` - [ ] Create a pull request in `twilio-java` - [ ] Provide a link below to the pull request, this ensures that the generated code has been verified - [ ] I have made a material change to the repo (functionality, testing, spelling, grammar) - [ ] I have read the [Contribution Guidelines](https://github.com/twilio/twilio-oai-generator/blob/main/CONTRIBUTING.md) and my PR follows them - [ ] I have titled the PR appropriately - [ ] I have updated my branch with the main branch - [ ] I have added tests that prove my fix is effective or that my feature works - [ ] I have added the necessary documentation about the functionality in the appropriate .md file - [ ] I have added inline documentation to the code I modified If you have questions, please create a GitHub Issue in this repository.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Fixes
Added token based pagination strategy
Checklist
If you have questions, please file a support ticket, or create a GitHub Issue in this repository.