Make v2 search limit hoisting explicit

.hark/changes/2026-10-31_owner_v2-search-limit-metadata.change.md

@@ -0,0 +1,7 @@
+---
+title: Keep custom search limits in the request body and use dedicated v2 search helpers
+semver_level: patch
+pr_url: https://github.com/stripe/stripe-python/pull/1960
+---
+
+Handwritten v2 search services should call `_request_v2_search` or `_request_v2_search_async` to send `limit` in the initial URL query. Ordinary requests no longer infer search behavior from the path.

stripe/_stripe_service.py

@@ -22,9 +22,8 @@ def _split_v2_search_params(
         return url, params
     body_params = dict(params)
     limit = body_params.pop("limit")
-    if "limit=" not in url:
-        separator = "&" if "?" in url else "?"
-        url = "%s%slimit=%s" % (url, separator, quote(str(limit), safe=""))
+    if "?" not in url:
+        url = "%s?limit=%s" % (url, quote(str(limit), safe=""))
     return url, body_params
 
 
@@ -46,12 +45,6 @@ def _request(
     ) -> StripeObject:
         if _param_encodings:
             params = _coerce_v2_params(params, _param_encodings)
-        if (
-            method == "post"
-            and url.split("?", 1)[0].startswith("/v2/")
-            and url.split("?", 1)[0].endswith("/search")
-        ):
-            url, params = _split_v2_search_params(url, params)
         return self._requestor.request(
             method,
             url,
@@ -73,12 +66,6 @@ async def _request_async(
     ) -> StripeObject:
         if _param_encodings:
             params = _coerce_v2_params(params, _param_encodings)
-        if (
-            method == "post"
-            and url.split("?", 1)[0].startswith("/v2/")
-            and url.split("?", 1)[0].endswith("/search")
-        ):
-            url, params = _split_v2_search_params(url, params)
         return await self._requestor.request_async(
             method,
             url,
@@ -88,6 +75,49 @@ async def _request_async(
             usage=["stripe_client"],
         )
 
+    # V2 search uses POST but sends limit in the URL rather than the JSON body.
+    # Add it only to the initial URL; page URLs contain their pagination parameters.
+    # Delegate to the ordinary request path exactly once.
+    def _request_v2_search(
+        self,
+        method: str,
+        url: str,
+        params: Optional[Mapping[str, Any]] = None,
+        options: Optional[RequestOptions] = None,
+        *,
+        base_address: BaseAddress,
+        _param_encodings: Optional[Dict[str, _SchemaNode]] = None,
+    ) -> StripeObject:
+        url, params = _split_v2_search_params(url, params)
+        return self._request(
+            method,
+            url,
+            params,
+            options,
+            base_address=base_address,
+            _param_encodings=_param_encodings,
+        )
+
+    async def _request_v2_search_async(
+        self,
+        method: str,
+        url: str,
+        params: Optional[Mapping[str, Any]] = None,
+        options: Optional[RequestOptions] = None,
+        *,
+        base_address: BaseAddress,
+        _param_encodings: Optional[Dict[str, _SchemaNode]] = None,
+    ) -> StripeObject:
+        url, params = _split_v2_search_params(url, params)
+        return await self._request_async(
+            method,
+            url,
+            params,
+            options,
+            base_address=base_address,
+            _param_encodings=_param_encodings,
+        )
+
     def _request_stream(
         self,
         method: str,

tests/test_v2_search_result_object.py

@@ -1,8 +1,165 @@
 import pytest
+from unittest.mock import AsyncMock, Mock
+
+from stripe._encode import _coerce_v2_params
+from stripe._stripe_service import StripeService
 
 from stripe.v2._search_result_object import SearchResultObject
 
 
+@pytest.mark.parametrize(
+    "url,params,expected_url,body",
+    [
+        (
+            "/v2/widgets/search",
+            {"limit": 2, "query": "widgets"},
+            "/v2/widgets/search?limit=2",
+            {"query": "widgets"},
+        ),
+        (
+            "/v2/widgets/search?page=2&limit=2",
+            {"limit": 9, "query": "widgets"},
+            "/v2/widgets/search?page=2&limit=2",
+            {"query": "widgets"},
+        ),
+        (
+            "/v2/widgets/search?page=2",
+            {"limit": 9, "query": "widgets"},
+            "/v2/widgets/search?page=2",
+            {"query": "widgets"},
+        ),
+        (
+            "/v2/widgets/search",
+            {"query": "widgets"},
+            "/v2/widgets/search",
+            {"query": "widgets"},
+        ),
+        ("/v2/widgets/search", None, "/v2/widgets/search", None),
+    ],
+)
+@pytest.mark.anyio
+async def test_search_helper_delegates_once(url, params, expected_url, body):
+    service = StripeService(Mock())
+    options = {"stripe_account": "acct_123"}
+    encodings = {"count": "int64_string"}
+    original = dict(params) if params is not None else None
+    service._request = Mock()
+    service._request_async = AsyncMock()
+    service._request_v2_search(
+        "post",
+        url,
+        params,
+        options,
+        base_address="api",
+        _param_encodings=encodings,
+    )
+    service._request.assert_called_once_with(
+        "post",
+        expected_url,
+        body,
+        options,
+        base_address="api",
+        _param_encodings=encodings,
+    )
+    await service._request_v2_search_async(
+        "post",
+        url,
+        params,
+        options,
+        base_address="api",
+        _param_encodings=encodings,
+    )
+    service._request_async.assert_awaited_once_with(
+        "post",
+        expected_url,
+        body,
+        options,
+        base_address="api",
+        _param_encodings=encodings,
+    )
+    assert params == original
+
+
+@pytest.mark.anyio
+async def test_ordinary_custom_search_keeps_limit_in_body():
+    requestor = Mock()
+    requestor.request_async = AsyncMock()
+    service = StripeService(requestor)
+    params = {"limit": 2, "query": "widgets"}
+    options = {"stripe_account": "acct_123"}
+    url = "/v2/data/operations/search"
+    service._request("post", url, params, options, base_address="api")
+    requestor.request.assert_called_once_with(
+        "post",
+        url,
+        params,
+        options,
+        base_address="api",
+        usage=["stripe_client"],
+    )
+    await service._request_async(
+        "post", url, params, options, base_address="api"
+    )
+    requestor.request_async.assert_awaited_once_with(
+        "post",
+        url,
+        params,
+        options,
+        base_address="api",
+        usage=["stripe_client"],
+    )
+    assert params == {"limit": 2, "query": "widgets"}
+
+
+@pytest.mark.anyio
+async def test_search_helper_preserves_encoding_options_and_usage(mocker):
+    coerce = mocker.patch(
+        "stripe._stripe_service._coerce_v2_params", wraps=_coerce_v2_params
+    )
+    requestor = Mock()
+    requestor.request_async = AsyncMock()
+    service = StripeService(requestor)
+    params = {"limit": 2, "count": 123}
+    options = {"stripe_account": "acct_123"}
+    encodings = {"count": "int64_string"}
+    service._request_v2_search(
+        "post",
+        "/v2/widgets/search",
+        params,
+        options,
+        base_address="meter_events",
+        _param_encodings=encodings,
+    )
+    coerce.assert_called_once_with({"count": 123}, encodings)
+    requestor.request.assert_called_once_with(
+        "post",
+        "/v2/widgets/search?limit=2",
+        {"count": "123"},
+        options,
+        base_address="meter_events",
+        usage=["stripe_client"],
+    )
+    coerce.reset_mock()
+    await service._request_v2_search_async(
+        "post",
+        "/v2/widgets/search",
+        params,
+        options,
+        base_address="meter_events",
+        _param_encodings=encodings,
+    )
+    coerce.assert_called_once_with({"count": 123}, encodings)
+    requestor.request_async.assert_awaited_once_with(
+        "post",
+        "/v2/widgets/search?limit=2",
+        {"count": "123"},
+        options,
+        base_address="meter_events",
+        usage=["stripe_client"],
+    )
+    assert params == {"limit": 2, "count": 123}
+
+
 class TestSearchResult(SearchResultObject):
     requests = []
     pages = []