feat: default requests to a 30 second timeout

Requests had no timeout, so a hung connection blocked the caller
indefinitely. niquests leaves `timeout` unset unless it is given.

Pass a 30 second `timeout` to the niquests session, matching the API's
own request timeout, and add a `timeout` option to `Seam` and
`SeamMultiWorkspace` so callers can raise or lower it. The option takes
the niquests forms: a number of seconds, a (connect, read) tuple, or
None for no timeout.

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

Author: claude

README.rst

@@ -65,6 +65,8 @@ Contents
 
     * `Setting the endpoint`_
 
+    * `Setting the request timeout`_
+
 * `Development and Testing`_
 
   * `Quickstart`_
@@ -436,6 +438,27 @@ e.g., testing or proxy setups.
 
 Either pass the ``endpoint`` option to the constructor, or set the ``SEAM_ENDPOINT`` environment variable.
 
+Setting the request timeout
+^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Requests time out after 30 seconds by default.
+Pass the ``timeout`` option, in seconds, to override this:
+
+.. code-block:: python
+
+    from seam import Seam
+
+    seam = Seam(api_key="your-api-key", timeout=60)
+
+The timeout may also be a ``(connect, read)`` tuple,
+and setting it to ``None`` disables the timeout entirely:
+
+.. code-block:: python
+
+    seam = Seam(api_key="your-api-key", timeout=(5, 60))
+
+A request that exceeds the timeout raises ``niquests.exceptions.Timeout``.
+
 Development and Testing
 -----------------------
 

seam/client.py

@@ -1,11 +1,11 @@
-from typing import Dict, Optional
+from typing import Dict, Optional, Tuple, Union
 from urllib.parse import urljoin
 import niquests as requests
 from importlib.metadata import version
 from urllib3.util import Retry
 import abc
 
-from .constants import LTS_VERSION
+from .constants import DEFAULT_TIMEOUT, LTS_VERSION
 from .exceptions import (
     SeamHttpApiError,
     SeamHttpInvalidInputError,
@@ -20,6 +20,8 @@
 
 DEFAULT_RETRIES = Retry()
 
+TimeoutType = Union[float, Tuple[float, float]]
+
 
 class AbstractSeamHttpClient(abc.ABC):
     @abc.abstractmethod
@@ -45,13 +47,18 @@ def __init__(
         base_url: str,
         auth_headers: Dict[str, str],
         retries: Optional[Retry] = DEFAULT_RETRIES,
+        timeout: Optional[TimeoutType] = DEFAULT_TIMEOUT,
         **kwargs
     ):
         # niquests.Session mounts its adapters while initializing, so retries
         # must be passed through here. Assigning self.retries afterwards leaves
         # the mounted adapters on their default and the option has no effect.
+        #
+        # timeout follows the niquests convention, where None means no timeout.
         super().__init__(
-            retries=DEFAULT_RETRIES if retries is None else retries, **kwargs
+            retries=DEFAULT_RETRIES if retries is None else retries,
+            timeout=timeout,
+            **kwargs
         )
 
         self.base_url = base_url

seam/constants.py

@@ -1,3 +1,5 @@
 LTS_VERSION = "1.0.0"
 
 DEFAULT_ENDPOINT = "https://connect.getseam.com"
+
+DEFAULT_TIMEOUT = 30

seam/seam.py

@@ -2,11 +2,11 @@
 from typing_extensions import Self
 from urllib3.util.retry import Retry
 
-from .constants import LTS_VERSION
+from .constants import DEFAULT_TIMEOUT, LTS_VERSION
 from .parse_options import parse_options
 from .routes import Routes
 from .models import AbstractSeam
-from .client import SeamHttpClient
+from .client import SeamHttpClient, TimeoutType
 from .paginator import SeamPaginator
 
 
@@ -42,6 +42,7 @@ def __init__(
         endpoint: Optional[str] = None,
         wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = True,
         retries: Optional[Retry] = None,
+        timeout: Optional[TimeoutType] = DEFAULT_TIMEOUT,
     ):
         """Initialize a Seam client instance.
 
@@ -66,6 +67,10 @@ def __init__(
         :type wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
         :param retries: Configuration for retry behavior on failed requests
         :type retries: Optional[urllib3.util.Retry]
+        :param timeout: The request timeout in seconds, or a
+            (connect, read) tuple. Defaults to 30 seconds. Pass None for no
+            timeout
+        :type timeout: Optional[Union[float, Tuple[float, float]]]
 
         :raises SeamInvalidOptionsError: If neither api_key nor
             personal_access_token is provided, or if workspace_id is missing
@@ -85,7 +90,10 @@ def __init__(
         self.defaults = {"wait_for_action_attempt": wait_for_action_attempt}
 
         self.client = SeamHttpClient(
-            base_url=endpoint, auth_headers=auth_headers, retries=retries
+            base_url=endpoint,
+            auth_headers=auth_headers,
+            retries=retries,
+            timeout=timeout,
         )
 
         Routes.__init__(self, client=self.client, defaults=self.defaults)
@@ -123,6 +131,7 @@ def from_api_key(
         endpoint: Optional[str] = None,
         wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = True,
         retries: Optional[Retry] = None,
+        timeout: Optional[TimeoutType] = DEFAULT_TIMEOUT,
     ) -> Self:
         """Create a Seam instance using an API key.
 
@@ -151,6 +160,7 @@ def from_api_key(
             endpoint=endpoint,
             wait_for_action_attempt=wait_for_action_attempt,
             retries=retries,
+            timeout=timeout,
         )
 
     @classmethod
@@ -162,6 +172,7 @@ def from_personal_access_token(
         endpoint: Optional[str] = None,
         wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = True,
         retries: Optional[Retry] = None,
+        timeout: Optional[TimeoutType] = DEFAULT_TIMEOUT,
     ) -> Self:
         """Create a Seam instance using a personal access token.
 
@@ -194,4 +205,5 @@ def from_personal_access_token(
             endpoint=endpoint,
             wait_for_action_attempt=wait_for_action_attempt,
             retries=retries,
+            timeout=timeout,
         )

seam/seam_multi_workspace.py

@@ -4,9 +4,9 @@
 from urllib3.util import Retry
 
 from .auth import get_auth_headers_for_multi_workspace_personal_access_token
-from .constants import LTS_VERSION
+from .constants import DEFAULT_TIMEOUT, LTS_VERSION
 from .options import get_endpoint
-from .client import SeamHttpClient
+from .client import SeamHttpClient, TimeoutType
 from .models import AbstractSeamMultiWorkspace
 from .routes.workspaces import Workspaces
 
@@ -52,6 +52,7 @@ def __init__(
         endpoint: Optional[str] = None,
         wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = True,
         retries: Optional[Retry] = None,
+        timeout: Optional[TimeoutType] = DEFAULT_TIMEOUT,
     ):
         """
         Initialize a SeamMultiWorkspace client instance.
@@ -71,6 +72,10 @@ def __init__(
         :type wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
         :param retries: Configuration for retry behavior on failed requests
         :type retries: Optional[urllib3.util.Retry]
+        :param timeout: The request timeout in seconds, or a
+            (connect, read) tuple. Defaults to 30 seconds. Pass None for no
+            timeout
+        :type timeout: Optional[Union[float, Tuple[float, float]]]
 
         :raises SeamInvalidTokenError: If the provided personal access token format is invalid
         """
@@ -86,6 +91,7 @@ def __init__(
             base_url=endpoint,
             auth_headers=auth_headers,
             retries=retries,
+            timeout=timeout,
         )
 
         defaults = {"wait_for_action_attempt": wait_for_action_attempt}
@@ -101,6 +107,7 @@ def from_personal_access_token(
         endpoint: Optional[str] = None,
         wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = True,
         retries: Optional[Retry] = None,
+        timeout: Optional[TimeoutType] = DEFAULT_TIMEOUT,
     ) -> Self:
         """
         Create a SeamMultiWorkspace instance using a personal access token.
@@ -132,4 +139,5 @@ def from_personal_access_token(
             endpoint=endpoint,
             wait_for_action_attempt=wait_for_action_attempt,
             retries=retries,
+            timeout=timeout,
         )

test/timeout_test.py

@@ -0,0 +1,81 @@
+import socket
+import threading
+from contextlib import contextmanager
+
+import niquests
+import pytest
+from urllib3.util import Retry
+
+from seam import Seam
+from seam.constants import DEFAULT_TIMEOUT
+
+
+def test_timeout_defaults_to_30_seconds():
+    seam = Seam.from_api_key("seam_apikey_token")
+
+    assert DEFAULT_TIMEOUT == 30
+    assert seam.client.timeout == 30
+
+
+def test_timeout_can_be_overridden():
+    seam = Seam.from_api_key("seam_apikey_token", timeout=60)
+
+    assert seam.client.timeout == 60
+
+
+def test_timeout_accepts_a_connect_read_tuple():
+    seam = Seam.from_api_key("seam_apikey_token", timeout=(5, 60))
+
+    assert seam.client.timeout == (5, 60)
+
+
+def test_timeout_can_be_disabled_with_none():
+    seam = Seam.from_api_key("seam_apikey_token", timeout=None)
+
+    assert seam.client.timeout is None
+
+
+def test_seam_times_out_a_request_that_never_responds():
+    with unresponsive_server() as endpoint:
+        seam = Seam.from_api_key(
+            "seam_apikey_token",
+            endpoint=endpoint,
+            timeout=0.25,
+            retries=Retry(total=0),
+        )
+
+        with pytest.raises(niquests.exceptions.Timeout):
+            seam.devices.list()
+
+
+@contextmanager
+def unresponsive_server():
+    """Accept connections but never send a response, so reads hang."""
+
+    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    listener.bind(("localhost", 0))
+    listener.listen(8)
+
+    accepted = []
+    stop = threading.Event()
+
+    def accept_forever():
+        while not stop.is_set():
+            try:
+                connection, _ = listener.accept()
+            except OSError:
+                return
+            accepted.append(connection)
+
+    thread = threading.Thread(target=accept_forever, daemon=True)
+    thread.start()
+
+    try:
+        yield f"http://localhost:{listener.getsockname()[1]}"
+    finally:
+        stop.set()
+        listener.close()
+        for connection in accepted:
+            connection.close()
+        thread.join(timeout=5)