mirror of
https://github.com/thousandeyes/thousandeyes-sdk-python.git
synced 2026-02-04 02:05:30 +00:00
Compare commits
2 Commits
abea9a244a
...
3182612069
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3182612069 | ||
|
|
3ad6e074bf |
@ -2,18 +2,28 @@
|
||||
|
||||
This package provides core functionality for interacting with the ThousandEyes API and should be installed before using any of the published SDKs.
|
||||
|
||||
Usage example for iterating paginated responses:
|
||||
`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 PaginatorIterator
|
||||
from thousandeyes_sdk.dashboards.api.dashboards_api import DashboardsApi
|
||||
from thousandeyes_sdk.core import Configuration, ApiClient, PaginatorIterator
|
||||
from thousandeyes_sdk.dashboards import DashboardsApi
|
||||
from itertools import islice
|
||||
|
||||
dashboards_api = DashboardsApi()
|
||||
for widget_data in PaginatorIterator(
|
||||
configuration = Configuration(
|
||||
host = "https://api.thousandeyes.com/v7",
|
||||
access_token = "an_access_token",
|
||||
)
|
||||
|
||||
|
||||
def get_dashboard_widget_data():
|
||||
with ApiClient(configuration) as client:
|
||||
dashboards_api = DashboardsApi(client)
|
||||
for item in list(islice(PaginatorIterator(
|
||||
dashboards_api.get_dashboard_widget_data,
|
||||
lambda response: response.data.tests if response.data else [],
|
||||
dashboard_id="dashboard-id",
|
||||
widget_id="widget-id",
|
||||
):
|
||||
print(widget_data)
|
||||
lambda response: response.data.tests,
|
||||
dashboard_id="a_dashboard_id",
|
||||
widget_id="a_widget_id",
|
||||
), 20)):
|
||||
print(item.test_id)
|
||||
```
|
||||
|
||||
@ -60,7 +60,8 @@ class PaginatorIterator(Generic[P, R, I]):
|
||||
|
||||
while True:
|
||||
response = self._method(**params)
|
||||
for item in self._items_getter(response):
|
||||
items = self._items_getter(response)
|
||||
for item in items if items else []:
|
||||
yield item
|
||||
|
||||
next_cursor = self._next_cursor_from_response(response)
|
||||
@ -71,14 +72,10 @@ class PaginatorIterator(Generic[P, R, I]):
|
||||
last_cursor = next_cursor
|
||||
|
||||
def _next_cursor_from_response(self, response: Any) -> Optional[str]:
|
||||
data = getattr(response, "data", response)
|
||||
links = getattr(data, "links", None)
|
||||
links = getattr(response, "links", None)
|
||||
|
||||
if links is None:
|
||||
links = getattr(data, "_links", None)
|
||||
|
||||
if links is None and isinstance(data, Mapping):
|
||||
links = data.get("_links") or data.get("links")
|
||||
links = getattr(response, "_links", None)
|
||||
|
||||
if links is None:
|
||||
return None
|
||||
|
||||
@ -14,8 +14,7 @@ def test_iterator_uses_cursor_from_next_href():
|
||||
else:
|
||||
links = SimpleNamespace(next=None)
|
||||
items = ["third"]
|
||||
data = SimpleNamespace(links=links)
|
||||
return SimpleNamespace(data=data, items=items)
|
||||
return SimpleNamespace(links=links, items=items)
|
||||
|
||||
responses = list(PaginatorIterator(method, lambda response: response.items))
|
||||
|
||||
@ -29,12 +28,12 @@ def test_iterator_reads_cursor_from_links_mapping():
|
||||
def method(**params):
|
||||
calls.append(params.copy())
|
||||
if params.get("pageCursor") is None:
|
||||
data = {"_links": {"next": {"href": "https://example.com?foo=1&pageCursor=xyz"}}}
|
||||
links = {"next": {"href": "https://example.com?foo=1&pageCursor=xyz"}}
|
||||
items = ["alpha"]
|
||||
else:
|
||||
data = {"_links": {"next": None}}
|
||||
links = {"next": None}
|
||||
items = ["beta"]
|
||||
return SimpleNamespace(data=data, items=items)
|
||||
return SimpleNamespace(links=links, items=items)
|
||||
|
||||
responses = list(PaginatorIterator(method, lambda response: response.items, cursor_param="pageCursor"))
|
||||
|
||||
@ -48,12 +47,12 @@ def test_iterator_stops_when_no_cursor_param_present():
|
||||
def method(**params):
|
||||
calls.append(params.copy())
|
||||
if params.get("cursor") is None:
|
||||
data = {"links": {"next": "/next/page"}}
|
||||
links = {"next": "/next/page"}
|
||||
items = ["one"]
|
||||
else:
|
||||
data = {"links": {"next": None}}
|
||||
links = {"next": None}
|
||||
items = ["two"]
|
||||
return SimpleNamespace(data=data, items=items)
|
||||
return SimpleNamespace(links=links, items=items)
|
||||
|
||||
responses = list(PaginatorIterator(method, lambda response: response.items))
|
||||
|
||||
@ -66,8 +65,8 @@ def test_iterator_stops_on_repeated_cursor():
|
||||
|
||||
def method(**params):
|
||||
calls.append(params.copy())
|
||||
data = {"links": {"next": "https://example.com?cursor=same"}}
|
||||
return SimpleNamespace(data=data, items=["only"])
|
||||
links = {"next": "https://example.com?cursor=same"}
|
||||
return SimpleNamespace(links=links, items=["only"])
|
||||
|
||||
responses = list(PaginatorIterator(method, lambda response: response.items, cursor="same"))
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user