From 8c37dadb51eedd03a95add9ae5d18bdc5322ea33 Mon Sep 17 00:00:00 2001 From: Rodrigo Rodrigues Date: Fri, 16 Jan 2026 17:09:05 +0000 Subject: [PATCH] CP-2451 Add paginator iterator helper --- thousandeyes-sdk-core/README.md | 15 +++ .../src/thousandeyes_sdk/core/__init__.py | 1 + .../core/pagination_iterator.py | 104 ++++++++++++++++++ .../test/test_pagination_iterator.py | 66 +++++++++++ 4 files changed, 186 insertions(+) create mode 100644 thousandeyes-sdk-core/src/thousandeyes_sdk/core/pagination_iterator.py create mode 100644 thousandeyes-sdk-core/test/test_pagination_iterator.py diff --git a/thousandeyes-sdk-core/README.md b/thousandeyes-sdk-core/README.md index 0999e656..66f1ddd4 100644 --- a/thousandeyes-sdk-core/README.md +++ b/thousandeyes-sdk-core/README.md @@ -1,3 +1,18 @@ # thousandeyes-sdk-core 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: + +```python +from thousandeyes_sdk.core import PaginatorIterator +from thousandeyes_sdk.usage.api.usage_api import UsageApi + +usage_api = UsageApi() +for page in PaginatorIterator( + usage_api.get_enterprise_agents_units_usage, + start_date="2024-01-01T00:00:00Z", + end_date="2024-01-31T23:59:59Z", +): + print(page) +``` diff --git a/thousandeyes-sdk-core/src/thousandeyes_sdk/core/__init__.py b/thousandeyes-sdk-core/src/thousandeyes_sdk/core/__init__.py index 9bba574f..8c6173c7 100644 --- a/thousandeyes-sdk-core/src/thousandeyes_sdk/core/__init__.py +++ b/thousandeyes-sdk-core/src/thousandeyes_sdk/core/__init__.py @@ -18,5 +18,6 @@ from . import exceptions from .api_client import ApiClient from .api_response import ApiResponse from .configuration import Configuration +from .pagination_iterator import PaginatorIterator import os.path diff --git a/thousandeyes-sdk-core/src/thousandeyes_sdk/core/pagination_iterator.py b/thousandeyes-sdk-core/src/thousandeyes_sdk/core/pagination_iterator.py new file mode 100644 index 00000000..57ca5613 --- /dev/null +++ b/thousandeyes-sdk-core/src/thousandeyes_sdk/core/pagination_iterator.py @@ -0,0 +1,104 @@ +# Copyright 2024 Cisco Systems, Inc. and its affiliates +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from typing import Any, Callable, Iterator, Mapping, Optional, TypeVar, Generic +from urllib.parse import parse_qs, urlparse + +from typing_extensions import ParamSpec + +P = ParamSpec("P") +R = TypeVar("R") + + +class PaginatorIterator(Generic[P, R]): + """Iterate over cursor-paginated responses. + + Calls ``method`` repeatedly, passing a cursor parameter between calls. + The next cursor is derived from ``response.data.links`` or ``response.data._links`` + (or mapping equivalents), supporting these link formats: + + - a direct ``href`` string + - a mapping with a ``href`` key + - an object with a ``href`` attribute + + Iteration stops when no next cursor is found or the cursor repeats. + """ + + def __init__( + self, + method: Callable[P, R], + *, + cursor_param: str = "cursor", + **params: P.kwargs, + ) -> None: + self._method = method + self._cursor_param = cursor_param + self._params: dict[str, Any] = dict(params) + + def __iter__(self) -> Iterator[R]: + params = dict(self._params) + last_cursor = params.get(self._cursor_param) + + while True: + response = self._method(**params) + yield response + + next_cursor = self._next_cursor_from_response(response) + if not next_cursor or next_cursor == last_cursor: + break + + params[self._cursor_param] = next_cursor + last_cursor = next_cursor + + def _next_cursor_from_response(self, response: Any) -> Optional[str]: + data = getattr(response, "data", response) + links = getattr(data, "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") + + if links is None: + return None + + next_link = getattr(links, "next", None) + if next_link is None and isinstance(links, Mapping): + next_link = links.get("next") + + if next_link is None: + return None + + if isinstance(next_link, str): + href = next_link + elif isinstance(next_link, Mapping): + href = next_link.get("href") + else: + href = getattr(next_link, "href", None) + + if not href: + return None + + parsed = urlparse(href) + query_params = parse_qs(parsed.query) + cursor_values = query_params.get(self._cursor_param) + + if cursor_values: + return cursor_values[0] + return None diff --git a/thousandeyes-sdk-core/test/test_pagination_iterator.py b/thousandeyes-sdk-core/test/test_pagination_iterator.py new file mode 100644 index 00000000..bb2da6cf --- /dev/null +++ b/thousandeyes-sdk-core/test/test_pagination_iterator.py @@ -0,0 +1,66 @@ +from types import SimpleNamespace + +from thousandeyes_sdk.core.iterable import PaginatorIterator + + +def test_iterator_uses_cursor_from_next_href(): + calls = [] + + def method(**params): + calls.append(params.copy()) + if params.get("cursor") is None: + links = SimpleNamespace(next="https://example.com/items?cursor=abc") + else: + links = SimpleNamespace(next=None) + data = SimpleNamespace(links=links) + return SimpleNamespace(data=data) + + responses = list(PaginatorIterator(method)) + + assert len(responses) == 2 + assert calls == [{}, {"cursor": "abc"}] + + +def test_iterator_reads_cursor_from_links_mapping(): + calls = [] + + def method(**params): + calls.append(params.copy()) + if params.get("pageCursor") is None: + data = {"_links": {"next": {"href": "https://example.com?foo=1&pageCursor=xyz"}}} + else: + data = {"_links": {"next": None}} + return SimpleNamespace(data=data) + + list(PaginatorIterator(method, cursor_param="pageCursor")) + + assert calls == [{}, {"pageCursor": "xyz"}] + + +def test_iterator_stops_when_no_cursor_param_present(): + calls = [] + + def method(**params): + calls.append(params.copy()) + if params.get("cursor") is None: + data = {"links": {"next": "/next/page"}} + else: + data = {"links": {"next": None}} + return SimpleNamespace(data=data) + + list(PaginatorIterator(method)) + + assert calls == [{}] + + +def test_iterator_stops_on_repeated_cursor(): + calls = [] + + def method(**params): + calls.append(params.copy()) + data = {"links": {"next": "https://example.com?cursor=same"}} + return SimpleNamespace(data=data) + + list(PaginatorIterator(method, cursor="same")) + + assert calls == [{"cursor": "same"}]