diff --git a/seam/modules/action_attempts.py b/seam/modules/action_attempts.py index 2c503e08..01fdd4a0 100644 --- a/seam/modules/action_attempts.py +++ b/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, diff --git a/seam/seam.py b/seam/seam.py index dfe59253..16b2f4d5 100644 --- a/seam/seam.py +++ b/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 diff --git a/seam/seam_without_workspace.py b/seam/seam_without_workspace.py index 9b3fb3aa..5857e24f 100644 --- a/seam/seam_without_workspace.py +++ b/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] diff --git a/test/wait_for_action_attempt_option_test.py b/test/wait_for_action_attempt_option_test.py new file mode 100644 index 00000000..43e76f30 --- /dev/null +++ b/test/wait_for_action_attempt_option_test.py @@ -0,0 +1,131 @@ +from typing import Any, cast + +import pytest + +from seam import AsyncSeam, Seam, SeamInvalidOptionsError + +JUNK_WAIT_VALUE = cast(Any, "true") + + +def test_constructor_rejects_a_non_bool_wait_for_action_attempt(): + with pytest.raises( + SeamInvalidOptionsError, + match='must be a bool or a dict with "timeout" and "polling_interval" keys, ' + "got str", + ): + Seam(api_key="seam_apikey_token", wait_for_action_attempt=JUNK_WAIT_VALUE) + + +def test_async_constructor_rejects_a_non_bool_wait_for_action_attempt(): + with pytest.raises( + SeamInvalidOptionsError, + match='must be a bool or a dict with "timeout" and "polling_interval" keys, ' + "got str", + ): + AsyncSeam(api_key="seam_apikey_token", wait_for_action_attempt=JUNK_WAIT_VALUE) + + +def test_constructor_rejects_an_unknown_wait_for_action_attempt_key(): + with pytest.raises( + SeamInvalidOptionsError, + match="got an unknown key 'poll_interval', " + 'expected "timeout" or "polling_interval"', + ): + Seam( + api_key="seam_apikey_token", + wait_for_action_attempt={cast(Any, "poll_interval"): 1}, + ) + + +def test_constructor_rejects_a_non_numeric_wait_for_action_attempt_value(): + with pytest.raises( + SeamInvalidOptionsError, + match="option 'timeout' must be a number, got str", + ): + Seam( + api_key="seam_apikey_token", + wait_for_action_attempt={"timeout": cast(Any, "5")}, + ) + + +def test_constructor_treats_none_as_the_default_wait_for_action_attempt(server): + endpoint, seed = server + seam = Seam( + api_key=seed["seam_apikey1_token"], + endpoint=endpoint, + wait_for_action_attempt=None, + ) + + assert seam.defaults["wait_for_action_attempt"] is True + + action_attempt = seam.locks.unlock_door(device_id=seed["august_device_1"]) + + assert action_attempt.status == "success" + + +def test_route_call_rejects_a_truthy_non_bool_wait_for_action_attempt(seam, server): + _, seed = server + + with pytest.raises( + SeamInvalidOptionsError, + match='must be a bool or a dict with "timeout" and "polling_interval" keys, ' + "got int", + ): + seam.locks.unlock_door( + device_id=seed["august_device_1"], + wait_for_action_attempt=cast(Any, 1), + ) + + +async def test_async_route_call_rejects_a_truthy_non_bool_wait_for_action_attempt( + async_seam, server +): + _, seed = server + + with pytest.raises( + SeamInvalidOptionsError, + match='must be a bool or a dict with "timeout" and "polling_interval" keys, ' + "got int", + ): + await async_seam.locks.unlock_door( + device_id=seed["august_device_1"], + wait_for_action_attempt=cast(Any, 1), + ) + + +def test_route_call_rejects_the_poll_interval_docstring_typo(seam, server): + _, seed = server + + with pytest.raises( + SeamInvalidOptionsError, + match="got an unknown key 'poll_interval'", + ): + seam.locks.unlock_door( + device_id=seed["august_device_1"], + wait_for_action_attempt={cast(Any, "poll_interval"): 1}, + ) + + +def test_wait_for_action_attempt_attribute_reflects_the_default(server): + endpoint, seed = server + seam = Seam(api_key=seed["seam_apikey1_token"], endpoint=endpoint) + + assert seam.wait_for_action_attempt is True + + seam.wait_for_action_attempt = False + + assert seam.defaults["wait_for_action_attempt"] is False + + action_attempt = seam.locks.unlock_door(device_id=seed["august_device_1"]) + + assert action_attempt.status == "pending" + + +def test_wait_for_action_attempt_attribute_rejects_junk(): + seam = Seam(api_key="seam_apikey_token") + + with pytest.raises( + SeamInvalidOptionsError, + match='must be a bool or a dict with "timeout" and "polling_interval" keys', + ): + seam.wait_for_action_attempt = JUNK_WAIT_VALUE