"""Base classes and the from_response factory for canonical errors.
Modeled on google-api-core's `GoogleAPICallError`:
- Two dimensions: HTTP status (transport) and canonical code (logical).
- A `_Retryable` marker mixin so retry policies can dispatch on type
instead of a hard-coded status set.
- A `from_response` factory that picks the most specific subclass
given an httpx.Response, preferring the structured `detail.code`
field when present.
The concrete subclasses live in sibling modules (one class per file)
and register themselves into the dispatch tables via
``_register_status`` / ``_register_code``.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, ClassVar
if TYPE_CHECKING:
import httpx
[docs]
class TaimoeError(Exception):
"""Base for every error raised by the Taimoe Platform SDK."""
class _Retryable:
"""Marker mixin: errors carrying this type are safe to retry.
Retry policies should branch on ``isinstance(exc, _Retryable)`` rather
than maintain their own list of retryable HTTP status codes.
"""
_STATUS_REGISTRY: dict[int, type["TaimoeAPIError"]] = {}
_CODE_REGISTRY: dict[str, type["TaimoeAPIError"]] = {}
def _register_status(status: int):
"""Decorator: register a subclass as the default for an HTTP status."""
def deco(cls: type["TaimoeAPIError"]) -> type["TaimoeAPIError"]:
_STATUS_REGISTRY[status] = cls
return cls
return deco
def _register_code(code: str):
"""Decorator: register a subclass for a canonical logical code.
Logical codes take precedence over HTTP status in `from_response`.
"""
def deco(cls: type["TaimoeAPIError"]) -> type["TaimoeAPIError"]:
_CODE_REGISTRY[code] = cls
return cls
return deco
[docs]
class TaimoeAPIError(TaimoeError):
"""Raised when an API request to the Taimoe Platform fails.
Attributes:
status_code: HTTP status from the response (or 0 for transport-level).
code: Canonical logical code (e.g. ``"RESOURCE_EXHAUSTED"``) if the
server returned a structured detail, else ``None``.
message: Human-readable message extracted from the response.
request_id: ``X-Request-ID`` echoed by the server, for log correlation.
details: Any additional fields from the structured detail body.
response: The raw httpx Response, kept for advanced debugging.
"""
#: Default canonical code for the subclass. Overridden by concrete classes.
default_code: ClassVar[str | None] = None
def __init__(
self,
message: str,
*,
status_code: int = 0,
code: str | None = None,
request_id: str | None = None,
details: dict[str, Any] | None = None,
response: "httpx.Response | None" = None,
) -> None:
super().__init__(message)
self.status_code = status_code
self.code = code or self.default_code
self.message = message
self.request_id = request_id
self.details = details or {}
self.response = response
def __str__(self) -> str:
parts: list[str] = []
if self.status_code:
parts.append(str(self.status_code))
if self.code:
parts.append(self.code)
head = " ".join(parts)
suffix = f" (request_id={self.request_id})" if self.request_id else ""
return f"{head}: {self.message}{suffix}" if head else f"{self.message}{suffix}"
[docs]
@classmethod
def from_response(cls, response: "httpx.Response") -> "TaimoeAPIError":
"""Build the most specific subclass for an HTTP error response.
Resolution order:
1. structured ``detail.code`` → ``_CODE_REGISTRY``
2. HTTP status → ``_STATUS_REGISTRY``
3. fall back to ``TaimoeAPIError``
"""
status = response.status_code
message, code, details = _parse_error_body(response)
request_id = response.headers.get("X-Request-ID")
target_cls: type[TaimoeAPIError]
if code and code in _CODE_REGISTRY:
target_cls = _CODE_REGISTRY[code]
elif status in _STATUS_REGISTRY:
target_cls = _STATUS_REGISTRY[status]
else:
target_cls = cls
return target_cls(
message=message,
status_code=status,
code=code,
request_id=request_id,
details=details,
response=response,
)
def _parse_error_body(
response: "httpx.Response",
) -> tuple[str, str | None, dict[str, Any]]:
"""Extract (message, canonical_code, extra_details) from an error body.
Handles three shapes:
1. ``{"detail": {"code": ..., "message": ..., ...}}`` — Taimoe canonical
2. ``{"detail": "..."}`` — FastAPI scalar form
3. anything else — fall back to ``response.text``
"""
fallback = response.text or f"HTTP {response.status_code}"
if not response.content:
return fallback, None, {}
try:
body = response.json()
except ValueError:
return fallback, None, {}
detail = body.get("detail") if isinstance(body, dict) else None
if isinstance(detail, dict):
message = detail.get("message") or fallback
code = detail.get("code")
extras = {k: v for k, v in detail.items() if k not in {"code", "message"}}
if "retry_after_seconds" not in extras:
retry_after = response.headers.get("Retry-After")
if retry_after is not None:
try:
extras["retry_after_seconds"] = int(retry_after)
except ValueError:
pass
return message, code, extras
if isinstance(detail, str):
return detail, None, {}
return fallback, None, {}