feat: add new token based pagination support

Author: kridai

tests/unit/base/test_token_pagination.py

@@ -1,7 +1,8 @@
 import unittest
-from unittest.mock import Mock, AsyncMock, MagicMock
+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
 
@@ -45,7 +46,10 @@ def setUp(self):
         """
         self.response.status_code = 200
 
-        self.solution = {"account_sid": "ACxxxx", "uri": "/Accounts/ACxxxx/Resources.json"}
+        self.solution = {
+            "account_sid": "ACxxxx",
+            "uri": "/Accounts/ACxxxx/Resources.json",
+        }
         self.page = TestTokenPaginationPage(self.version, self.response, self.solution)
 
     def test_key_property(self):
@@ -282,7 +286,7 @@ def test_next_page_without_uri_in_solution(self):
 
         page = TestTokenPaginationPage(version, response, solution)
 
-        with self.assertRaises(ValueError) as context:
+        with self.assertRaises(TwilioException) as context:
             page.next_page()
 
         self.assertIn("URI must be provided", str(context.exception))
@@ -308,7 +312,7 @@ def test_previous_page_without_uri_in_solution(self):
 
         page = TestTokenPaginationPage(version, response, solution)
 
-        with self.assertRaises(ValueError) as context:
+        with self.assertRaises(TwilioException) as context:
             page.previous_page()
 
         self.assertIn("URI must be provided", str(context.exception))
@@ -481,7 +485,9 @@ def setUp(self):
         )
 
         self.version = self.client.api.v2010
-        self.response = self.version.page(method="GET", uri="/Accounts/ACaaaa/Records.json")
+        self.response = self.version.page(
+            method="GET", uri="/Accounts/ACaaaa/Records.json"
+        )
 
         self.solution = {
             "account_sid": "ACaaaa",

tests/unit/rest/test_client.py

@@ -81,23 +81,32 @@ def test_periods_in_query(self):
     def test_edge_deprecation_warning_when_only_edge_is_set(self):
         with warnings.catch_warnings(record=True) as w:
             warnings.simplefilter("always")  # Ensure all warnings are caught
-            Client(username="username", password="password", edge="edge")  # Trigger the warning
+            Client(
+                username="username", password="password", edge="edge"
+            )  # Trigger the warning
 
             # Check if a warning was raised
             self.assertGreater(len(w), 0)
             self.assertTrue(issubclass(w[-1].category, DeprecationWarning))
-            self.assertIn("For regional processing, DNS is of format product.<edge>.<region>.twilio.com; otherwise use product.twilio.com.", str(w[-1].message))
+            self.assertIn(
+                "For regional processing, DNS is of format product.<edge>.<region>.twilio.com; otherwise use product.twilio.com.",
+                str(w[-1].message),
+            )
 
     def test_edge_deprecation_warning_when_only_region_is_set(self):
         with warnings.catch_warnings(record=True) as w:
             warnings.simplefilter("always")  # Ensure all warnings are caught
-            Client(username="username", password="password", region="us1")  # Trigger the warning
+            Client(
+                username="username", password="password", region="us1"
+            )  # Trigger the warning
 
             # Check if a warning was raised
             self.assertGreater(len(w), 0)
             self.assertTrue(issubclass(w[-1].category, DeprecationWarning))
-            self.assertIn("For regional processing, DNS is of format product.<edge>.<region>.twilio.com; otherwise use product.twilio.com.", str(w[-1].message))
-
+            self.assertIn(
+                "For regional processing, DNS is of format product.<edge>.<region>.twilio.com; otherwise use product.twilio.com.",
+                str(w[-1].message),
+            )
 
 
 class TestUserAgentClients(unittest.TestCase):

twilio/base/client_base.py

@@ -13,6 +13,7 @@
 
 class ClientBase(object):
     """A client for accessing the Twilio API."""
+
     region_mappings = {
         "au1": "sydney",
         "br1": "sao-paulo",
@@ -22,8 +23,9 @@ class ClientBase(object):
         "jp2": "osaka",
         "sg1": "singapore",
         "us1": "ashburn",
-        "us2": "umatilla"
+        "us2": "umatilla",
     }
+
     def __init__(
         self,
         username: Optional[str] = None,
@@ -56,13 +58,19 @@ def __init__(
         """ :type : str """
         self.password = password or environment.get("TWILIO_AUTH_TOKEN")
         """ :type : str """
-        if (edge is not None and region is None) or (region is not None and edge is None):
+        if (edge is not None and region is None) or (
+            region is not None and edge is None
+        ):
             warnings.warn(
                 "For regional processing, DNS is of format product.<edge>.<region>.twilio.com; otherwise use product.twilio.com.",
                 DeprecationWarning,
-                stacklevel=2
+                stacklevel=2,
             )
-        self.edge = edge or environment.get("TWILIO_EDGE") or (self.region_mappings[region] if region is not None else "")
+        self.edge = (
+            edge
+            or environment.get("TWILIO_EDGE")
+            or (self.region_mappings[region] if region is not None else "")
+        )
         """ :type : str """
         self.region = region or environment.get("TWILIO_REGION")
         """ :type : str """

twilio/base/token_pagination.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Optional
+from typing import Optional
 
 from twilio.base.exceptions import TwilioException
 from twilio.base.page import Page
@@ -85,7 +85,9 @@ def _get_page(self, token: Optional[str]) -> Optional["TokenPagination"]:
         cls = type(self)
         return cls(self._version, response, self._solution)
 
-    async def _get_page_async(self, token: Optional[str]) -> Optional["TokenPagination"]:
+    async def _get_page_async(
+        self, token: Optional[str]
+    ) -> Optional["TokenPagination"]:
         """
         Internal async helper to fetch a page using a given token.
 
@@ -104,7 +106,9 @@ async def _get_page_async(self, token: Optional[str]) -> Optional["TokenPaginati
             raise TwilioException("URI must be provided for token pagination")
 
         url = self._version.domain.absolute_url(uri)
-        response = await self._version.domain.twilio.request_async("GET", url, params=params)
+        response = await self._version.domain.twilio.request_async(
+            "GET", url, params=params
+        )
         cls = type(self)
         return cls(self._version, response, self._solution)