mirror of
https://github.com/thousandeyes/thousandeyes-sdk-python.git
synced 2026-02-04 10:56:51 +00:00
Compare commits
5 Commits
02441092cf
...
8303751a45
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8303751a45 | ||
|
|
3182612069 | ||
|
|
3ad6e074bf | ||
|
|
abea9a244a | ||
|
|
8c37dadb51 |
@ -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.
|
||||
|
||||
`PaginationIterable` is unbounded, so wrap it with `itertools.islice` to cap the number of items and avoid making unintended, potentially expensive API calls.
|
||||
`PaginatorIterator` 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:
|
||||
|
||||
```python
|
||||
from thousandeyes_sdk.core import Configuration, ApiClient, PaginationIterable
|
||||
from thousandeyes_sdk.core import Configuration, ApiClient, PaginatorIterator
|
||||
from thousandeyes_sdk.dashboards import DashboardsApi
|
||||
from itertools import islice
|
||||
|
||||
@ -19,7 +19,7 @@ configuration = Configuration(
|
||||
def get_dashboard_widget_data():
|
||||
with ApiClient(configuration) as client:
|
||||
dashboards_api = DashboardsApi(client)
|
||||
for item in list(islice(PaginationIterable(
|
||||
for item in list(islice(PaginatorIterator(
|
||||
dashboards_api.get_dashboard_widget_data,
|
||||
lambda response: response.data.tests,
|
||||
dashboard_id="a_dashboard_id",
|
||||
|
||||
@ -18,6 +18,6 @@ from . import exceptions
|
||||
from .api_client import ApiClient
|
||||
from .api_response import ApiResponse
|
||||
from .configuration import Configuration
|
||||
from .pagination_iterable import PaginationIterable
|
||||
from .pagination_iterator import PaginatorIterator
|
||||
|
||||
import os.path
|
||||
|
||||
@ -26,7 +26,7 @@ P = ParamSpec("P")
|
||||
R = TypeVar("R")
|
||||
I = TypeVar("I")
|
||||
|
||||
class PaginationIterable(Generic[P, R, I]):
|
||||
class PaginatorIterator(Generic[P, R, I]):
|
||||
"""Iterate over cursor-paginated responses.
|
||||
|
||||
Calls ``method`` repeatedly, passing a cursor parameter between calls,
|
||||
@ -1,9 +1,9 @@
|
||||
from types import SimpleNamespace
|
||||
|
||||
from thousandeyes_sdk.core.pagination_iterable import PaginationIterable
|
||||
from thousandeyes_sdk.core.pagination_iterator import PaginatorIterator
|
||||
|
||||
|
||||
def test_iterable_uses_cursor_from_next_href():
|
||||
def test_iterator_uses_cursor_from_next_href():
|
||||
calls = []
|
||||
|
||||
def method(**params):
|
||||
@ -16,13 +16,13 @@ def test_iterable_uses_cursor_from_next_href():
|
||||
items = ["third"]
|
||||
return SimpleNamespace(links=links, items=items)
|
||||
|
||||
responses = list(PaginationIterable(method, lambda response: response.items))
|
||||
responses = list(PaginatorIterator(method, lambda response: response.items))
|
||||
|
||||
assert responses == ["first", "second", "third"]
|
||||
assert calls == [{}, {"cursor": "abc"}]
|
||||
|
||||
|
||||
def test_iterable_reads_cursor_from_links_mapping():
|
||||
def test_iterator_reads_cursor_from_links_mapping():
|
||||
calls = []
|
||||
|
||||
def method(**params):
|
||||
@ -35,13 +35,13 @@ def test_iterable_reads_cursor_from_links_mapping():
|
||||
items = ["beta"]
|
||||
return SimpleNamespace(links=links, items=items)
|
||||
|
||||
responses = list(PaginationIterable(method, lambda response: response.items, cursor_param="pageCursor"))
|
||||
responses = list(PaginatorIterator(method, lambda response: response.items, cursor_param="pageCursor"))
|
||||
|
||||
assert responses == ["alpha", "beta"]
|
||||
assert calls == [{}, {"pageCursor": "xyz"}]
|
||||
|
||||
|
||||
def test_iterable_stops_when_no_cursor_param_present():
|
||||
def test_iterator_stops_when_no_cursor_param_present():
|
||||
calls = []
|
||||
|
||||
def method(**params):
|
||||
@ -54,13 +54,13 @@ def test_iterable_stops_when_no_cursor_param_present():
|
||||
items = ["two"]
|
||||
return SimpleNamespace(links=links, items=items)
|
||||
|
||||
responses = list(PaginationIterable(method, lambda response: response.items))
|
||||
responses = list(PaginatorIterator(method, lambda response: response.items))
|
||||
|
||||
assert responses == ["one"]
|
||||
assert calls == [{}]
|
||||
|
||||
|
||||
def test_iterable_stops_on_repeated_cursor():
|
||||
def test_iterator_stops_on_repeated_cursor():
|
||||
calls = []
|
||||
|
||||
def method(**params):
|
||||
@ -68,7 +68,7 @@ def test_iterable_stops_on_repeated_cursor():
|
||||
links = {"next": "https://example.com?cursor=same"}
|
||||
return SimpleNamespace(links=links, items=["only"])
|
||||
|
||||
responses = list(PaginationIterable(method, lambda response: response.items, cursor="same"))
|
||||
responses = list(PaginatorIterator(method, lambda response: response.items, cursor="same"))
|
||||
|
||||
assert responses == ["only"]
|
||||
assert calls == [{"cursor": "same"}]
|
||||
Loading…
Reference in New Issue
Block a user