feat: make response parsing total and surface unknown action attempt statuses

Seam adds event types, action types, and error codes between SDK releases. A
payload this version does not recognize should stay readable rather than cost
the caller the whole response.

Reading a response no longer fails on the shape of the payload. Shared helpers
in seam/parse.py guard every conversion the generated dataclasses perform, so a
list sent as a scalar reads as empty, a nested object sent as a scalar reads as
None, an unrecognized discriminator yields a DeepAttrDict, and a known variant
whose payload does not convert degrades to one too. A malformed free-form record
now degrades on its own rather than costing the surrounding resource its
generated type.

Waiting on an action attempt whose status is neither pending, success, nor error
raises the new SeamActionAttemptUnknownStatusError instead of returning it as a
success. Reporting success would tell the caller the action completed when the
SDK cannot tell, and the previous behavior did exactly that. The error subclasses
SeamActionAttemptError, so existing handlers for that base keep working. Reading
the status no longer assumes the attempt carries one, which used to raise
AttributeError on a fallback payload.

DeepAttrDict is re-exported from the package root so callers can test for the
fallback against the public API.

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

Author: claude

codegen/layouts/partials/resource-dataclass.hbs

@@ -31,8 +31,10 @@
 {{#unless properties}}
 {{memberIndent}}    # pylint: disable=unused-argument
 {{/unless}}
+{{memberIndent}}    if not isinstance(d, dict):
+{{memberIndent}}        d = {}
 {{memberIndent}}    return cls(
 {{#each properties}}
-{{../memberIndent}}        {{pythonIdentifier name}}={{#if isRequiredObject}}cls.{{nestedClassName}}.from_dict(d.get("{{name}}") or {}){{else}}{{#if isObject}}cls.{{nestedClassName}}.from_dict(d.get("{{name}}")) if d.get("{{name}}") is not None else None{{else}}{{#if isDiscriminatedObjectList}}[_from_discriminated_dict(i, cls._{{nestedClassName}}Variants, "{{discriminator}}") for i in d.get("{{name}}") or []]{{else}}{{#if isObjectList}}[cls.{{nestedClassName}}.from_dict(i) for i in d.get("{{name}}") or []]{{else}}{{#if isDictParam}}DeepAttrDict({{/if}}d.get("{{name}}", None){{#if isDictParam}}){{/if}}{{/if}}{{/if}}{{/if}}{{/if}},
+{{../memberIndent}}        {{pythonIdentifier name}}={{#if isRequiredObject}}_required_object_from_dict(cls.{{nestedClassName}}, d.get("{{name}}")){{else}}{{#if isObject}}_object_from_dict(cls.{{nestedClassName}}, d.get("{{name}}")){{else}}{{#if isDiscriminatedObjectList}}_discriminated_list_from_dict(d.get("{{name}}"), cls._{{nestedClassName}}Variants, "{{discriminator}}"){{else}}{{#if isObjectList}}_object_list_from_dict(cls.{{nestedClassName}}, d.get("{{name}}")){{else}}{{#if isDictParam}}_record_from_dict({{/if}}d.get("{{name}}", None){{#if isDictParam}}){{/if}}{{/if}}{{/if}}{{/if}}{{/if}},
 {{/each}}
 {{memberIndent}}    )

codegen/layouts/resource.hbs

@@ -1,17 +1,15 @@
 from typing import Any, Dict, List, Literal, Optional, {{#if union.secondaryDiscriminator}}Tuple, {{/if}}Union{{#if union}}, cast{{/if}}
 from dataclasses import dataclass
 from ..deep_attr_dict import DeepAttrDict
+from ..parse import (
+    discriminated_list_from_dict as _discriminated_list_from_dict,
+    object_from_dict as _object_from_dict,
+    object_list_from_dict as _object_list_from_dict,
+    record_from_dict as _record_from_dict,
+    required_object_from_dict as _required_object_from_dict,
+)
 from ..resource_mapping import ResourceMapping
 
-{{#if hasDiscriminatedLists}}
-
-def _from_discriminated_dict(
-    d: Any, variants: Dict[str, Any], discriminator: str
-) -> Any:
-    variant = variants.get(d.get(discriminator))
-    return DeepAttrDict(d) if variant is None else variant.from_dict(d)
-{{/if}}
-
 {{#each classes}}
 {{> resource-dataclass}}
 
@@ -36,11 +34,28 @@ def _from_discriminated_dict(
 def {{union.fromDictName}}(d: Any) -> {{union.className}}:
     """Deserialize a known {{union.discriminator}}{{#if union.secondaryDiscriminator}} and {{union.secondaryDiscriminator}}{{/if}} variant.
 
-    Unknown discriminator values return ``DeepAttrDict`` so payloads from a
-    newer API remain readable. The static return type covers known variants.
+    An unrecognized discriminator, or a known one whose payload does not
+    convert, returns ``DeepAttrDict`` so payloads from a newer API remain
+    readable. The static return type covers known variants.
     """
-    variant = {{union.variantsName}}.get({{#if union.secondaryDiscriminator}}(d.get("{{union.discriminator}}"), d.get("{{union.secondaryDiscriminator}}")){{else}}d.get("{{union.discriminator}}"){{/if}})
+    if not isinstance(d, dict):
+        return cast({{union.className}}, DeepAttrDict(d) if isinstance(d, dict) else d)
+{{#if union.secondaryDiscriminator}}
+    key = (d.get("{{union.discriminator}}"), d.get("{{union.secondaryDiscriminator}}"))
+    variant = (
+        {{union.variantsName}}.get(cast(Tuple[str, str], key))
+        if isinstance(key[0], str) and isinstance(key[1], str)
+        else None
+    )
+{{else}}
+    key = d.get("{{union.discriminator}}")
+    variant = {{union.variantsName}}.get(key) if isinstance(key, str) else None
+{{/if}}
     if variant is None:
         return cast({{union.className}}, DeepAttrDict(d))
-    return variant.from_dict(d)
+    try:
+        return variant.from_dict(d)
+    except Exception:  # pylint: disable=broad-exception-caught
+        # A known discriminator whose payload does not convert is still readable.
+        return cast({{union.className}}, DeepAttrDict(d))
 {{/if}}

seam/__init__.py

@@ -16,7 +16,9 @@
     SeamActionAttemptError,
     SeamActionAttemptFailedError,
     SeamActionAttemptTimeoutError,
+    SeamActionAttemptUnknownStatusError,
 )
+from .deep_attr_dict import DeepAttrDict
 from .seam_webhook import SeamWebhook
 from svix.webhooks import WebhookVerificationError as SeamWebhookVerificationError
 from .null import NULL, Null

seam/exceptions.py

@@ -254,3 +254,36 @@ def __init__(self, action_attempt: PendingActionAttempt, timeout: float):
         message = f"Timed out waiting for action attempt after {timeout}s"
         super().__init__(message, action_attempt)
         self.name = self.__class__.__name__
+
+
+class SeamActionAttemptUnknownStatusError(SeamActionAttemptError):
+    """
+    Exception raised when an action attempt reports a status this SDK does not know.
+
+    Seam adds action attempt statuses over time. Waiting for an attempt promises to
+    return a succeeded attempt or raise, and an unrecognized status supports neither
+    conclusion: reporting success would claim the action completed when the SDK
+    cannot tell, and polling on would block until the timeout and then report a
+    misleading timeout. Read ``action_attempt`` to inspect the status directly.
+
+    :ivar name: Name of the exception class
+    :vartype name: str
+    :ivar status: The unrecognized status reported by the API
+    :vartype status: str
+    """
+
+    def __init__(self, action_attempt: ActionAttempt, status: str):
+        """
+        :param action_attempt: The ActionAttempt object carrying the unknown status
+        :type action_attempt: ActionAttempt
+        :param status: The unrecognized status reported by the API
+        :type status: str
+        """
+
+        message = (
+            f'Action attempt reported an unknown status "{status}". '
+            "This SDK version may predate it; upgrade or read the action attempt directly."
+        )
+        super().__init__(message, action_attempt)
+        self.name = self.__class__.__name__
+        self.status = status

seam/modules/action_attempts.py

@@ -1,11 +1,21 @@
-from typing import Dict, Optional, Union
+from typing import Dict, Optional, Union, cast
 import asyncio
 import time
 
 from ..client import AsyncSeamHttpClient, SeamHttpClient
-from ..exceptions import SeamActionAttemptFailedError, SeamActionAttemptTimeoutError
+from ..exceptions import (
+    SeamActionAttemptFailedError,
+    SeamActionAttemptTimeoutError,
+    SeamActionAttemptUnknownStatusError,
+)
 from ..options import SeamInvalidOptionsError
-from ..resources import ActionAttempt, SuccessActionAttempt, action_attempt_from_dict
+from ..resources import (
+    ActionAttempt,
+    ErrorActionAttempt,
+    PendingActionAttempt,
+    SuccessActionAttempt,
+    action_attempt_from_dict,
+)
 from ..response import unwrap
 
 TIMEOUT = 5.0
@@ -14,6 +24,16 @@
 WAIT_FOR_ACTION_ATTEMPT_OPTION_KEYS = ("timeout", "polling_interval")
 
 
+def _status_of(action_attempt: ActionAttempt) -> Optional[str]:
+    """Read the status without assuming the attempt carries one.
+
+    An attempt from a newer API deserializes to a DeepAttrDict, where a missing
+    key raises rather than reading as None.
+    """
+
+    return getattr(action_attempt, "status", None)
+
+
 def validate_wait_for_action_attempt(
     value: Optional[Union[bool, Dict[str, float]]],
 ) -> None:
@@ -93,20 +113,28 @@ def poll_until_ready(
     if action_attempt is None:
         action_attempt = get_action_attempt(client, action_attempt_id)
 
-    while action_attempt.status == "pending":
+    while _status_of(action_attempt) == "pending":
         remaining = deadline - time.monotonic()
 
         if remaining <= 0:
-            raise SeamActionAttemptTimeoutError(action_attempt, timeout)
+            raise SeamActionAttemptTimeoutError(
+                cast(PendingActionAttempt, action_attempt), timeout
+            )
 
         time.sleep(min(polling_interval, remaining))
 
         action_attempt = get_action_attempt(client, action_attempt_id)
 
-    if action_attempt.status == "error":
-        raise SeamActionAttemptFailedError(action_attempt)
+    if _status_of(action_attempt) == "error":
+        raise SeamActionAttemptFailedError(cast(ErrorActionAttempt, action_attempt))
 
-    return action_attempt
+    # Neither pending, success, nor error: a status added after this SDK release.
+    # Waiting cannot conclude, so say so rather than pass it off as a success.
+    status = _status_of(action_attempt)
+    if status != "success":
+        raise SeamActionAttemptUnknownStatusError(action_attempt, str(status))
+
+    return cast(SuccessActionAttempt, action_attempt)
 
 
 def resolve_action_attempt(
@@ -165,20 +193,28 @@ async def poll_until_ready_async(
     if action_attempt is None:
         action_attempt = await get_action_attempt_async(client, action_attempt_id)
 
-    while action_attempt.status == "pending":
+    while _status_of(action_attempt) == "pending":
         remaining = deadline - time.monotonic()
 
         if remaining <= 0:
-            raise SeamActionAttemptTimeoutError(action_attempt, timeout)
+            raise SeamActionAttemptTimeoutError(
+                cast(PendingActionAttempt, action_attempt), timeout
+            )
 
         await asyncio.sleep(min(polling_interval, remaining))
 
         action_attempt = await get_action_attempt_async(client, action_attempt_id)
 
-    if action_attempt.status == "error":
-        raise SeamActionAttemptFailedError(action_attempt)
+    if _status_of(action_attempt) == "error":
+        raise SeamActionAttemptFailedError(cast(ErrorActionAttempt, action_attempt))
 
-    return action_attempt
+    # Neither pending, success, nor error: a status added after this SDK release.
+    # Waiting cannot conclude, so say so rather than pass it off as a success.
+    status = _status_of(action_attempt)
+    if status != "success":
+        raise SeamActionAttemptUnknownStatusError(action_attempt, str(status))
+
+    return cast(SuccessActionAttempt, action_attempt)
 
 
 async def resolve_action_attempt_async(

seam/parse.py

@@ -0,0 +1,99 @@
+"""Total conversion helpers shared by the generated resource dataclasses.
+
+The Seam API grows new event types, action types, and error codes between SDK
+releases, so reading a response must never fail on the shape of the payload.
+Every helper here degrades instead of raising: an unusable value becomes
+``None``, an empty list, or a :class:`DeepAttrDict` carrying the payload
+verbatim, so one unexpected field cannot cost the caller the whole response.
+
+The narrow cost is that a genuinely malformed payload is no longer loud. The
+payload survives on the returned object, so it stays diagnosable.
+"""
+
+from typing import Any, Dict, List
+
+from .deep_attr_dict import DeepAttrDict
+
+
+def record_from_dict(value: Any) -> Any:
+    """Wrap a free-form record for attribute access, passing anything else through.
+
+    A record the API sends as something other than an object degrades to the raw
+    value rather than costing the caller the whole surrounding resource.
+    """
+
+    return DeepAttrDict(value) if isinstance(value, dict) else value
+
+
+def object_from_dict(cls: Any, value: Any) -> Any:
+    """Convert an optional nested object, or None when it is unusable."""
+
+    if not isinstance(value, dict):
+        return None
+
+    try:
+        return cls.from_dict(value)
+    except Exception:  # pylint: disable=broad-exception-caught
+        return DeepAttrDict(value)
+
+
+def required_object_from_dict(cls: Any, value: Any) -> Any:
+    """Convert a nested object the caller may always read, never returning None."""
+
+    source = value if isinstance(value, dict) else {}
+
+    try:
+        return cls.from_dict(source)
+    except Exception:  # pylint: disable=broad-exception-caught
+        return DeepAttrDict(source)
+
+
+def object_list_from_dict(cls: Any, value: Any) -> List[Any]:
+    """Convert a list of nested objects, skipping nothing and raising for nothing.
+
+    A value that is not a list reads as empty, so callers can always iterate.
+    """
+
+    if not isinstance(value, list):
+        return []
+
+    return [
+        object_from_dict(cls, item) if isinstance(item, dict) else item
+        for item in value
+    ]
+
+
+def discriminated_from_dict(
+    value: Any, variants: Dict[str, Any], discriminator: str
+) -> Any:
+    """Convert one member of a discriminated union.
+
+    An unrecognized discriminator, or a known one whose payload does not
+    convert, yields a DeepAttrDict so a newer API stays readable.
+    """
+
+    if not isinstance(value, dict):
+        # Not an object at all; hand it back untouched rather than lose it.
+        return value
+
+    key = value.get(discriminator)
+    variant = variants.get(key) if isinstance(key, str) else None
+
+    if variant is None:
+        return DeepAttrDict(value)
+
+    try:
+        return variant.from_dict(value)
+    except Exception:  # pylint: disable=broad-exception-caught
+        return DeepAttrDict(value)
+
+
+def discriminated_list_from_dict(
+    value: Any, variants: Dict[str, Any], discriminator: str
+) -> List[Any]:
+    """Convert a list of discriminated union members, degrading item by item."""
+
+    if not isinstance(value, list):
+        return []
+
+    return [discriminated_from_dict(item, variants, discriminator) for item in value]