Skip to content

fix: Validate the wait_for_action_attempt option #637

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
razor-x merged 1 commit into from
Aug 28, 2026
+264 −20
Merged

fix: Validate the wait_for_action_attempt option #637

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 47 additions & 0 deletions seam/modules/action_attempts.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
53 changes: 43 additions & 10 deletions seam/seam.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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]
Expand All @@ -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,
Expand All @@ -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:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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]
Expand All @@ -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,
Expand All @@ -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:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
53 changes: 43 additions & 10 deletions seam/seam_without_workspace.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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]
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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]
Expand Down Expand Up @@ -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]
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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]
Expand Down
Loading
Loading