fix: Validate the wait_for_action_attempt option

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

Author: claude

seam/modules/action_attempts.py

@@ -10,6 +10,49 @@
 TIMEOUT = 5.0
 POLLING_INTERVAL = 0.5
 
+WAIT_FOR_ACTION_ATTEMPT_OPTION_KEYS = ("timeout", "polling_interval")
+
+
+def validate_wait_for_action_attempt(
+    value: Optional[Union[bool, Dict[str, float]]],
+) -> None:
+    if isinstance(value, bool):
+        return
+
+    if isinstance(value, dict):
+        for key, option in value.items():
+            if key not in WAIT_FOR_ACTION_ATTEMPT_OPTION_KEYS:
+                raise SeamInvalidOptionsError(
+                    f"The wait_for_action_attempt option got an unknown key {key!r}, "
+                    'expected "timeout" or "polling_interval"'
+                )
+
+            if isinstance(option, bool) or not isinstance(option, (int, float)):
+                raise SeamInvalidOptionsError(
+                    f"The wait_for_action_attempt option {key!r} must be a number, "
+                    f"got {type(option).__name__}"
+                )
+
+        return
+
+    raise SeamInvalidOptionsError(
+        "The wait_for_action_attempt option must be a bool or a dict with "
+        f'"timeout" and "polling_interval" keys, got {type(value).__name__}'
+    )
+
+
+def normalize_wait_for_action_attempt(
+    value: Optional[Union[bool, Dict[str, float]]],
+) -> Union[bool, Dict[str, float]]:
+    """Resolve None to the default and reject anything but a bool or options dict."""
+
+    if value is None:
+        return True
+
+    validate_wait_for_action_attempt(value)
+
+    return value
+
 
 def validate_poll_options(timeout: float, polling_interval: float) -> None:
     # Written as negated comparisons so NaN fails both checks.
@@ -69,6 +112,8 @@ def resolve_action_attempt(
     action_attempt: ActionAttempt,
     wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]],
 ) -> ActionAttempt:
+    validate_wait_for_action_attempt(wait_for_action_attempt)
+
     if wait_for_action_attempt is True:
         return poll_until_ready(
             client=client,
@@ -137,6 +182,8 @@ async def resolve_action_attempt_async(
     action_attempt: ActionAttempt,
     wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]],
 ) -> ActionAttempt:
+    validate_wait_for_action_attempt(wait_for_action_attempt)
+
     if wait_for_action_attempt is True:
         return await poll_until_ready_async(
             client=client,

seam/seam.py

@@ -7,6 +7,7 @@
 from .routes import AsyncRoutes, Routes
 from .models import AbstractAsyncSeam, AbstractSeam
 from .client import AsyncSeamHttpClient, SeamHttpClient
+from .modules.action_attempts import normalize_wait_for_action_attempt
 from .paginator import AsyncSeamPaginator, SeamPaginator
 
 
@@ -63,7 +64,7 @@ def __init__(
         :type endpoint: Optional[str]
         :param wait_for_action_attempt: Controls whether to wait for an
             action attempt to complete. Can be a boolean or a dictionary with
-            'timeout' and 'poll_interval' keys
+            'timeout' and 'polling_interval' keys
         :type wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
         :param retries: Configuration for retry behavior on failed requests
         :type retries: Optional[httpx_retries.Retry]
@@ -81,14 +82,17 @@ def __init__(
             access token format is invalid
         """
 
-        self.wait_for_action_attempt = wait_for_action_attempt
         auth_headers, endpoint = parse_options(
             api_key=api_key,
             personal_access_token=personal_access_token,
             workspace_id=workspace_id,
             endpoint=endpoint,
         )
-        self.defaults = {"wait_for_action_attempt": wait_for_action_attempt}
+        self.defaults = {
+            "wait_for_action_attempt": normalize_wait_for_action_attempt(
+                wait_for_action_attempt
+            )
+        }
 
         self.client = SeamHttpClient(
             base_url=endpoint,
@@ -103,6 +107,19 @@ def __init__(
         # namespaces passes a self the signature does not admit.
         Routes.__init__(self, client=self.client, defaults=self.defaults)  # type: ignore[arg-type]
 
+    @property
+    def wait_for_action_attempt(self) -> Union[bool, Dict[str, float]]:
+        """Default wait behavior for action attempts, shared with every route."""
+        return self.defaults["wait_for_action_attempt"]
+
+    @wait_for_action_attempt.setter
+    def wait_for_action_attempt(
+        self, value: Optional[Union[bool, Dict[str, float]]]
+    ) -> None:
+        self.defaults["wait_for_action_attempt"] = normalize_wait_for_action_attempt(
+            value
+        )
+
     def create_paginator(
         self, request: Callable, params: Optional[Dict[str, Any]] = None, /
     ) -> SeamPaginator:
@@ -173,7 +190,7 @@ def from_api_key(
         :type endpoint: Optional[str]
         :param wait_for_action_attempt: Controls whether to wait for an
             action attempt to complete. Can be a boolean or a dictionary with
-            'timeout' and 'poll_interval' keys
+            'timeout' and 'polling_interval' keys
         :type wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
         :return: A new instance of the Seam class authenticated with the
             provided API key
@@ -219,7 +236,7 @@ def from_personal_access_token(
         :type endpoint: Optional[str]
         :param wait_for_action_attempt: Controls whether to wait for an
             action attempt to complete. Can be a boolean or a dictionary with
-            'timeout' and 'poll_interval' keys
+            'timeout' and 'polling_interval' keys
         :type wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
         :return: A new instance of the Seam class authenticated with the
             provided personal access token
@@ -294,7 +311,7 @@ def __init__(
         :type endpoint: Optional[str]
         :param wait_for_action_attempt: Controls whether to wait for an
             action attempt to complete. Can be a boolean or a dictionary with
-            'timeout' and 'poll_interval' keys
+            'timeout' and 'polling_interval' keys
         :type wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
         :param retries: Configuration for retry behavior on failed requests
         :type retries: Optional[httpx_retries.Retry]
@@ -312,14 +329,17 @@ def __init__(
             access token format is invalid
         """
 
-        self.wait_for_action_attempt = wait_for_action_attempt
         auth_headers, endpoint = parse_options(
             api_key=api_key,
             personal_access_token=personal_access_token,
             workspace_id=workspace_id,
             endpoint=endpoint,
         )
-        self.defaults = {"wait_for_action_attempt": wait_for_action_attempt}
+        self.defaults = {
+            "wait_for_action_attempt": normalize_wait_for_action_attempt(
+                wait_for_action_attempt
+            )
+        }
 
         self.client = AsyncSeamHttpClient(
             base_url=endpoint,
@@ -335,6 +355,19 @@ def __init__(
         # admit.
         AsyncRoutes.__init__(self, client=self.client, defaults=self.defaults)  # type: ignore[arg-type]
 
+    @property
+    def wait_for_action_attempt(self) -> Union[bool, Dict[str, float]]:
+        """Default wait behavior for action attempts, shared with every route."""
+        return self.defaults["wait_for_action_attempt"]
+
+    @wait_for_action_attempt.setter
+    def wait_for_action_attempt(
+        self, value: Optional[Union[bool, Dict[str, float]]]
+    ) -> None:
+        self.defaults["wait_for_action_attempt"] = normalize_wait_for_action_attempt(
+            value
+        )
+
     def create_paginator(
         self, request: Callable, params: Optional[Dict[str, Any]] = None, /
     ) -> AsyncSeamPaginator:
@@ -402,7 +435,7 @@ def from_api_key(
         :type endpoint: Optional[str]
         :param wait_for_action_attempt: Controls whether to wait for an
             action attempt to complete. Can be a boolean or a dictionary with
-            'timeout' and 'poll_interval' keys
+            'timeout' and 'polling_interval' keys
         :type wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
         :return: A new instance of the AsyncSeam class authenticated with the
             provided API key
@@ -445,7 +478,7 @@ def from_personal_access_token(
         :type endpoint: Optional[str]
         :param wait_for_action_attempt: Controls whether to wait for an
             action attempt to complete. Can be a boolean or a dictionary with
-            'timeout' and 'poll_interval' keys
+            'timeout' and 'polling_interval' keys
         :type wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
         :return: A new instance of the AsyncSeam class authenticated with the
             provided personal access token

seam/seam_without_workspace.py

@@ -6,6 +6,7 @@
 from .parse_options import parse_without_workspace_options
 from .client import AsyncSeamHttpClient, SeamHttpClient
 from .models import AbstractAsyncSeamWithoutWorkspace, AbstractSeamWithoutWorkspace
+from .modules.action_attempts import normalize_wait_for_action_attempt
 from .routes.workspaces import AsyncWorkspaces, Workspaces
 
 
@@ -63,7 +64,7 @@ def __init__(
         :type endpoint: Optional[str]
         :param wait_for_action_attempt: Controls whether to wait for an
             action attempt to complete. Can be a boolean or a dictionary with
-            'timeout' and 'poll_interval' keys
+            'timeout' and 'polling_interval' keys
         :type wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
         :param retries: Configuration for retry behavior on failed requests
         :type retries: Optional[httpx_retries.Retry]
@@ -79,7 +80,6 @@ def __init__(
         :raises SeamInvalidTokenError: If the provided personal access token format is invalid
         """
 
-        self.wait_for_action_attempt = wait_for_action_attempt
         auth_headers, endpoint = parse_without_workspace_options(
             personal_access_token=personal_access_token,
             endpoint=endpoint,
@@ -93,11 +93,28 @@ def __init__(
             httpx_options=httpx_options,
         )
 
-        defaults = {"wait_for_action_attempt": wait_for_action_attempt}
+        self.defaults = {
+            "wait_for_action_attempt": normalize_wait_for_action_attempt(
+                wait_for_action_attempt
+            )
+        }
 
-        self._workspaces = Workspaces(client=self.client, defaults=defaults)
+        self._workspaces = Workspaces(client=self.client, defaults=self.defaults)
         self.workspaces = WorkspacesProxy(self._workspaces)
 
+    @property
+    def wait_for_action_attempt(self) -> Union[bool, Dict[str, float]]:
+        """Default wait behavior for action attempts, shared with every route."""
+        return self.defaults["wait_for_action_attempt"]
+
+    @wait_for_action_attempt.setter
+    def wait_for_action_attempt(
+        self, value: Optional[Union[bool, Dict[str, float]]]
+    ) -> None:
+        self.defaults["wait_for_action_attempt"] = normalize_wait_for_action_attempt(
+            value
+        )
+
     @classmethod
     def from_personal_access_token(
         cls,
@@ -121,7 +138,7 @@ def from_personal_access_token(
         :type endpoint: Optional[str]
         :param wait_for_action_attempt: Controls whether to wait for an
             action attempt to complete. Can be a boolean or a dictionary with
-            'timeout' and 'poll_interval' keys
+            'timeout' and 'polling_interval' keys
         :type wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
         :param retries: Configuration for retry behavior on failed requests
         :type retries: Optional[httpx_retries.Retry]
@@ -209,7 +226,7 @@ def __init__(
         :type endpoint: Optional[str]
         :param wait_for_action_attempt: Controls whether to wait for an
             action attempt to complete. Can be a boolean or a dictionary with
-            'timeout' and 'poll_interval' keys
+            'timeout' and 'polling_interval' keys
         :type wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
         :param retries: Configuration for retry behavior on failed requests
         :type retries: Optional[httpx_retries.Retry]
@@ -225,7 +242,6 @@ def __init__(
         :raises SeamInvalidTokenError: If the provided personal access token format is invalid
         """
 
-        self.wait_for_action_attempt = wait_for_action_attempt
         auth_headers, endpoint = parse_without_workspace_options(
             personal_access_token=personal_access_token,
             endpoint=endpoint,
@@ -239,11 +255,28 @@ def __init__(
             httpx_options=httpx_options,
         )
 
-        defaults = {"wait_for_action_attempt": wait_for_action_attempt}
+        self.defaults = {
+            "wait_for_action_attempt": normalize_wait_for_action_attempt(
+                wait_for_action_attempt
+            )
+        }
 
-        self._workspaces = AsyncWorkspaces(client=self.client, defaults=defaults)
+        self._workspaces = AsyncWorkspaces(client=self.client, defaults=self.defaults)
         self.workspaces = AsyncWorkspacesProxy(self._workspaces)
 
+    @property
+    def wait_for_action_attempt(self) -> Union[bool, Dict[str, float]]:
+        """Default wait behavior for action attempts, shared with every route."""
+        return self.defaults["wait_for_action_attempt"]
+
+    @wait_for_action_attempt.setter
+    def wait_for_action_attempt(
+        self, value: Optional[Union[bool, Dict[str, float]]]
+    ) -> None:
+        self.defaults["wait_for_action_attempt"] = normalize_wait_for_action_attempt(
+            value
+        )
+
     @classmethod
     def from_personal_access_token(
         cls,
@@ -264,7 +297,7 @@ def from_personal_access_token(
         :type endpoint: Optional[str]
         :param wait_for_action_attempt: Controls whether to wait for an
             action attempt to complete. Can be a boolean or a dictionary with
-            'timeout' and 'poll_interval' keys
+            'timeout' and 'polling_interval' keys
         :type wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
         :param retries: Configuration for retry behavior on failed requests
         :type retries: Optional[httpx_retries.Retry]