fix: Reject the retries option with a custom transport

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y1RzepycXEYA3LStfjt8cY

Author: claude

README.rst

@@ -682,6 +682,31 @@ and its ``Retry`` class is re-exported from ``seam`` for convenience:
         retries=Retry(total=3, backoff_factor=0.5, status_forcelist=[503]),
     )
 
+Bringing your own transport
++++++++++++++++++++++++++++
+
+A custom ``transport`` or ``mounts`` passed through ``httpx_options`` replaces
+the transport the SDK builds, so it takes full responsibility for retries:
+requests through it are not retried unless you wrap it yourself. Combining
+either with the ``retries`` option raises a ``SeamInvalidOptionsError``. To
+retry through your own transport, wrap it with ``RetryTransport``:
+
+.. code-block:: python
+
+    from httpx_retries import RetryTransport
+
+    from seam import Seam, Retry
+
+    seam = Seam(
+        api_key="your-api-key",
+        httpx_options={
+            "transport": RetryTransport(
+                transport=MyCustomTransport(),
+                retry=Retry(total=2, status_forcelist=[429, 503]),
+            ),
+        },
+    )
+
 Configuring the httpx client
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

seam/client.py

@@ -14,6 +14,7 @@
     SeamHttpUnauthorizedError,
 )
 from .null import replace_null
+from .options import SeamInvalidOptionsError
 from .strict_url_search_params_serializer import serialize_url_search_params
 
 SDK_HEADERS = {
@@ -105,14 +106,24 @@ def __init__(
         self,
         base_url: str,
         auth_headers: Dict[str, str],
-        retries: Optional[Retry] = DEFAULT_RETRIES,
+        retries: Optional[Retry] = None,
         timeout: Optional[float] = DEFAULT_TIMEOUT,
         httpx_options: Optional[Dict[str, Any]] = None,
         **kwargs,
     ):
         options = _build_client_options(base_url, timeout, httpx_options, kwargs)
 
         custom_headers = options.pop("headers", {})
+
+        if retries is not None and (
+            options.get("transport") is not None or options.get("mounts") is not None
+        ):
+            raise SeamInvalidOptionsError(
+                "The retries option cannot be combined with a custom transport "
+                "or mounts, which bypass the retry transport; wrap your "
+                "transport with httpx_retries.RetryTransport instead"
+            )
+
         self._retry_policy = DEFAULT_RETRIES if retries is None else retries
 
         super().__init__(**options)
@@ -170,14 +181,24 @@ def __init__(
         self,
         base_url: str,
         auth_headers: Dict[str, str],
-        retries: Optional[Retry] = DEFAULT_RETRIES,
+        retries: Optional[Retry] = None,
         timeout: Optional[float] = DEFAULT_TIMEOUT,
         httpx_options: Optional[Dict[str, Any]] = None,
         **kwargs,
     ):
         options = _build_client_options(base_url, timeout, httpx_options, kwargs)
 
         custom_headers = options.pop("headers", {})
+
+        if retries is not None and (
+            options.get("transport") is not None or options.get("mounts") is not None
+        ):
+            raise SeamInvalidOptionsError(
+                "The retries option cannot be combined with a custom transport "
+                "or mounts, which bypass the retry transport; wrap your "
+                "transport with httpx_retries.RetryTransport instead"
+            )
+
         self._retry_policy = DEFAULT_RETRIES if retries is None else retries
 
         super().__init__(**options)

test/custom_transport_test.py

@@ -0,0 +1,81 @@
+import httpx
+import pytest
+from httpx_retries import Retry, RetryTransport
+
+from seam import AsyncSeam, Seam, SeamHttpApiError, SeamInvalidOptionsError
+
+CONFLICT_MESSAGE = (
+    "The retries option cannot be combined with a custom transport or mounts"
+)
+
+
+def test_retries_with_a_custom_transport_raises():
+    with pytest.raises(SeamInvalidOptionsError, match=CONFLICT_MESSAGE):
+        Seam(
+            api_key="seam_apikey_token",
+            retries=Retry(total=3),
+            httpx_options={"transport": httpx.HTTPTransport()},
+        )
+
+
+def test_retries_with_custom_mounts_raises():
+    with pytest.raises(SeamInvalidOptionsError, match=CONFLICT_MESSAGE):
+        Seam(
+            api_key="seam_apikey_token",
+            retries=Retry(total=3),
+            httpx_options={
+                "mounts": {"https://": httpx.HTTPTransport()},
+            },
+        )
+
+
+def test_retries_with_a_custom_transport_raises_async():
+    with pytest.raises(SeamInvalidOptionsError, match=CONFLICT_MESSAGE):
+        AsyncSeam(
+            api_key="seam_apikey_token",
+            retries=Retry(total=3),
+            httpx_options={"transport": httpx.AsyncHTTPTransport()},
+        )
+
+
+def test_a_custom_transport_is_not_retried(recording_server):
+    with recording_server(
+        [
+            (503, {"error": {"type": "service_unavailable", "message": "Down"}}),
+            (200, {"device": {"device_id": "x"}}),
+        ]
+    ) as (endpoint, requests):
+        seam = Seam.from_api_key(
+            "seam_apikey_token",
+            endpoint=endpoint,
+            httpx_options={"transport": httpx.HTTPTransport()},
+        )
+
+        with pytest.raises(SeamHttpApiError):
+            seam.devices.get(device_id="x")
+
+        assert len(requests) == 1
+
+
+def test_a_wrapped_custom_transport_retries(recording_server):
+    with recording_server(
+        [
+            (503, {"error": {"type": "service_unavailable", "message": "Down"}}),
+            (200, {"device": {"device_id": "x"}}),
+        ]
+    ) as (endpoint, requests):
+        seam = Seam.from_api_key(
+            "seam_apikey_token",
+            endpoint=endpoint,
+            httpx_options={
+                "transport": RetryTransport(
+                    transport=httpx.HTTPTransport(),
+                    retry=Retry(total=2, status_forcelist=[503]),
+                ),
+            },
+        )
+
+        device = seam.devices.get(device_id="x")
+
+        assert device.device_id == "x"
+        assert len(requests) == 2