Compare commits

..

2 Commits

Author SHA1 Message Date
Rodrigo Rodrigues
02441092cf
Merge 7b833c5ea7 into c5916a3b66 2026-01-26 09:24:32 +00:00
Rodrigo Rodrigues
7b833c5ea7 feat: Add pagination iterable helper 2026-01-26 09:24:20 +00:00
4 changed files with 14 additions and 14 deletions

View File

@ -2,11 +2,11 @@
This package provides core functionality for interacting with the ThousandEyes API and should be installed before using any of the published SDKs. This package provides core functionality for interacting with the ThousandEyes API and should be installed before using any of the published SDKs.
`PaginatorIterator` is unbounded, so wrap it with `itertools.islice` to cap the number of items and avoid making unintended, potentially expensive API calls. `PaginationIterable` is unbounded, so wrap it with `itertools.islice` to cap the number of items and avoid making unintended, potentially expensive API calls.
Pick a slice size that matches your UI or batch size so you only fetch what you plan to process: Pick a slice size that matches your UI or batch size so you only fetch what you plan to process:
```python ```python
from thousandeyes_sdk.core import Configuration, ApiClient, PaginatorIterator from thousandeyes_sdk.core import Configuration, ApiClient, PaginationIterable
from thousandeyes_sdk.dashboards import DashboardsApi from thousandeyes_sdk.dashboards import DashboardsApi
from itertools import islice from itertools import islice
@ -19,7 +19,7 @@ configuration = Configuration(
def get_dashboard_widget_data(): def get_dashboard_widget_data():
with ApiClient(configuration) as client: with ApiClient(configuration) as client:
dashboards_api = DashboardsApi(client) dashboards_api = DashboardsApi(client)
for item in list(islice(PaginatorIterator( for item in list(islice(PaginationIterable(
dashboards_api.get_dashboard_widget_data, dashboards_api.get_dashboard_widget_data,
lambda response: response.data.tests, lambda response: response.data.tests,
dashboard_id="a_dashboard_id", dashboard_id="a_dashboard_id",

View File

@ -18,6 +18,6 @@ from . import exceptions
from .api_client import ApiClient from .api_client import ApiClient
from .api_response import ApiResponse from .api_response import ApiResponse
from .configuration import Configuration from .configuration import Configuration
from .pagination_iterator import PaginatorIterator from .pagination_iterable import PaginationIterable
import os.path import os.path

View File

@ -26,7 +26,7 @@ P = ParamSpec("P")
R = TypeVar("R") R = TypeVar("R")
I = TypeVar("I") I = TypeVar("I")
class PaginatorIterator(Generic[P, R, I]): class PaginationIterable(Generic[P, R, I]):
"""Iterate over cursor-paginated responses. """Iterate over cursor-paginated responses.
Calls ``method`` repeatedly, passing a cursor parameter between calls, Calls ``method`` repeatedly, passing a cursor parameter between calls,

View File

@ -1,9 +1,9 @@
from types import SimpleNamespace from types import SimpleNamespace
from thousandeyes_sdk.core.pagination_iterator import PaginatorIterator from thousandeyes_sdk.core.pagination_iterable import PaginationIterable
def test_iterator_uses_cursor_from_next_href(): def test_iterable_uses_cursor_from_next_href():
calls = [] calls = []
def method(**params): def method(**params):
@ -16,13 +16,13 @@ def test_iterator_uses_cursor_from_next_href():
items = ["third"] items = ["third"]
return SimpleNamespace(links=links, items=items) return SimpleNamespace(links=links, items=items)
responses = list(PaginatorIterator(method, lambda response: response.items)) responses = list(PaginationIterable(method, lambda response: response.items))
assert responses == ["first", "second", "third"] assert responses == ["first", "second", "third"]
assert calls == [{}, {"cursor": "abc"}] assert calls == [{}, {"cursor": "abc"}]
def test_iterator_reads_cursor_from_links_mapping(): def test_iterable_reads_cursor_from_links_mapping():
calls = [] calls = []
def method(**params): def method(**params):
@ -35,13 +35,13 @@ def test_iterator_reads_cursor_from_links_mapping():
items = ["beta"] items = ["beta"]
return SimpleNamespace(links=links, items=items) return SimpleNamespace(links=links, items=items)
responses = list(PaginatorIterator(method, lambda response: response.items, cursor_param="pageCursor")) responses = list(PaginationIterable(method, lambda response: response.items, cursor_param="pageCursor"))
assert responses == ["alpha", "beta"] assert responses == ["alpha", "beta"]
assert calls == [{}, {"pageCursor": "xyz"}] assert calls == [{}, {"pageCursor": "xyz"}]
def test_iterator_stops_when_no_cursor_param_present(): def test_iterable_stops_when_no_cursor_param_present():
calls = [] calls = []
def method(**params): def method(**params):
@ -54,13 +54,13 @@ def test_iterator_stops_when_no_cursor_param_present():
items = ["two"] items = ["two"]
return SimpleNamespace(links=links, items=items) return SimpleNamespace(links=links, items=items)
responses = list(PaginatorIterator(method, lambda response: response.items)) responses = list(PaginationIterable(method, lambda response: response.items))
assert responses == ["one"] assert responses == ["one"]
assert calls == [{}] assert calls == [{}]
def test_iterator_stops_on_repeated_cursor(): def test_iterable_stops_on_repeated_cursor():
calls = [] calls = []
def method(**params): def method(**params):
@ -68,7 +68,7 @@ def test_iterator_stops_on_repeated_cursor():
links = {"next": "https://example.com?cursor=same"} links = {"next": "https://example.com?cursor=same"}
return SimpleNamespace(links=links, items=["only"]) return SimpleNamespace(links=links, items=["only"])
responses = list(PaginatorIterator(method, lambda response: response.items, cursor="same")) responses = list(PaginationIterable(method, lambda response: response.items, cursor="same"))
assert responses == ["only"] assert responses == ["only"]
assert calls == [{"cursor": "same"}] assert calls == [{"cursor": "same"}]