From 46941682172469c8e198e297259f3dd4b371fb90 Mon Sep 17 00:00:00 2001 From: jar Date: Fri, 2 Oct 2026 22:10:03 +0000 Subject: [PATCH] Add API v2 search result pagination Committed-By-Agent: goose Orbit-Session-Id: ac7fd08a-e385-4498-800e-21f51d3e4ad6 --- stripe/_object_classes.py | 4 ++ stripe/_util.py | 1 + stripe/v2/__init__.py | 1 + stripe/v2/_search_result_object.py | 67 +++++++++++++++++++++++++ tests/test_v2_search_result_object.py | 71 +++++++++++++++++++++++++++ 5 files changed, 144 insertions(+) create mode 100644 stripe/v2/_search_result_object.py create mode 100644 tests/test_v2_search_result_object.py diff --git a/stripe/_object_classes.py b/stripe/_object_classes.py index 377c69906..3f7679509 100644 --- a/stripe/_object_classes.py +++ b/stripe/_object_classes.py @@ -344,6 +344,10 @@ } V2_OBJECT_CLASSES: Dict[str, Tuple[str, str]] = { + "v2.search_result": ( + "stripe.v2._search_result_object", + "SearchResultObject", + ), # V2 Object classes: The beginning of the section generated from our OpenAPI spec "v2.billing.meter_event": ("stripe.v2.billing._meter_event", "MeterEvent"), "v2.billing.meter_event_adjustment": ( diff --git a/stripe/_util.py b/stripe/_util.py index 8b378c63c..e034b209f 100644 --- a/stripe/_util.py +++ b/stripe/_util.py @@ -288,6 +288,7 @@ def _convert_to_stripe_object( and ( (getattr(obj, "object") == "list") or (getattr(obj, "object") == "search_result") + or (getattr(obj, "object") == "v2.search_result") ) ): obj._retrieve_params = params diff --git a/stripe/v2/__init__.py b/stripe/v2/__init__.py index 3eb03dfd8..2df834066 100644 --- a/stripe/v2/__init__.py +++ b/stripe/v2/__init__.py @@ -1,5 +1,6 @@ from typing_extensions import TYPE_CHECKING from stripe.v2._list_object import ListObject as ListObject +from stripe.v2._search_result_object import SearchResultObject as SearchResultObject from stripe.v2._amount import Amount as Amount, AmountParam as AmountParam diff --git a/stripe/v2/_search_result_object.py b/stripe/v2/_search_result_object.py new file mode 100644 index 000000000..662aec9d0 --- /dev/null +++ b/stripe/v2/_search_result_object.py @@ -0,0 +1,67 @@ +from copy import deepcopy +from typing import ( + AsyncIterator, + Generic, + Iterator, + List, + Mapping, + Any, + Optional, + TypeVar, +) + +from stripe._any_iterator import AnyIterator +from stripe._stripe_object import StripeObject + + +T = TypeVar("T", bound=StripeObject) + + +class SearchResultObject(StripeObject, Generic[T]): + """A page of API v2 search results with POST-based auto-pagination.""" + + OBJECT_NAME = "v2.search_result" + data: List[T] + next_page_url: Optional[str] + previous_page_url: Optional[str] + total_count: int + + def __iter__(self) -> Iterator[T]: + return getattr(self, "data", []).__iter__() + + def __len__(self) -> int: + return getattr(self, "data", []).__len__() + + def auto_paging_iter(self) -> AnyIterator[T]: + return AnyIterator(self._auto_paging_iter(), self._auto_paging_iter_async()) + + def _original_params(self) -> Mapping[str, Any]: + return deepcopy(self._retrieve_params) + + def _auto_paging_iter(self) -> Iterator[T]: + page: SearchResultObject[T] = self + params = self._original_params() + while True: + for item in page.data: + yield item + if page.next_page_url is None: + break + result = self._request( + "post", page.next_page_url, params=params, base_address="api" + ) + assert isinstance(result, SearchResultObject) + page = result + + async def _auto_paging_iter_async(self) -> AsyncIterator[T]: + page: SearchResultObject[T] = self + params = self._original_params() + while True: + for item in page.data: + yield item + if page.next_page_url is None: + break + result = await self._request_async( + "post", page.next_page_url, params=params, base_address="api" + ) + assert isinstance(result, SearchResultObject) + page = result diff --git a/tests/test_v2_search_result_object.py b/tests/test_v2_search_result_object.py new file mode 100644 index 000000000..071600c80 --- /dev/null +++ b/tests/test_v2_search_result_object.py @@ -0,0 +1,71 @@ +import pytest + +from stripe.v2 import SearchResultObject + + +class TestSearchResult(SearchResultObject): + requests = [] + pages = [] + + def _request(self, method, url, params=None, **kwargs): + self.requests.append((method, url, params, kwargs)) + return self.pages.pop(0) + + async def _request_async(self, method, url, params=None, **kwargs): + return self._request(method, url, params=params, **kwargs) + + +def make_result(data, next_page_url): + result = TestSearchResult._construct_from( + values={ + "object": "v2.search_result", + "data": data, + "next_page_url": next_page_url, + "previous_page_url": None, + "total_count": 2, + }, + last_response=None, + requestor=None, + api_mode="V2", + ) + return result + + +def test_auto_paging_replays_original_post_body(): + params = { + "query": 'status:"active"', + "sort": ["name", "-created"], + "limit": 2, + "future_field": {"enabled": True}, + } + first = make_result(["one"], "/v2/widgets/search?page=2") + first._retrieve_params = params + TestSearchResult.requests = [] + TestSearchResult.pages = [ + make_result([], "/v2/widgets/search?page=3"), + make_result(["two"], None), + ] + + assert list(first.auto_paging_iter()) == ["one", "two"] + assert [ + (method, url, body) for method, url, body, _ in TestSearchResult.requests + ] == [ + ("post", "/v2/widgets/search?page=2", params), + ("post", "/v2/widgets/search?page=3", params), + ] + + +@pytest.mark.asyncio +async def test_async_auto_paging_replays_original_post_body(): + first = make_result(["one"], "/v2/widgets/search?page=2") + first._retrieve_params = {"query": "widgets"} + TestSearchResult.requests = [] + TestSearchResult.pages = [make_result(["two"], None)] + + results = [item async for item in first.auto_paging_iter()] + assert results == ["one", "two"] + assert TestSearchResult.requests[0][:3] == ( + "post", + "/v2/widgets/search?page=2", + {"query": "widgets"}, + )