Spaces:
Runtime error
Runtime error
File size: 3,163 Bytes
46252cd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | """Typed error hierarchy for the OpenWA Python SDK.
The OpenWA API returns NestJS-default errors of the shape::
{"statusCode": int, "message": str | list[str], "error": str}
This module maps that to a typed, ergonomic error tree so callers can
``isinstance``-check or branch on ``.status``.
"""
from __future__ import annotations
from typing import Any
class OpenWAError(Exception):
"""Base class for every error raised by the SDK."""
class OpenWAApiError(OpenWAError):
"""Raised when the API responds with a non-2xx status.
Attributes:
status: HTTP status code.
body: Parsed JSON body if available, otherwise the raw text.
error_kind: Value of the ``error`` field in the NestJS envelope.
"""
def __init__(self, message: str, status: int, body: Any = None, error_kind: str | None = None) -> None:
super().__init__(message)
self.status = status
self.body = body
self.error_kind = error_kind
@classmethod
def from_response(cls, status_code: int, text: str, context: str) -> "OpenWAApiError":
import json
body: Any = None
if text:
try:
body = json.loads(text)
except ValueError:
body = text
envelope = body if isinstance(body, dict) and "statusCode" in body else None
raw_message = envelope.get("message") if envelope else body
if isinstance(raw_message, list):
message_text = ", ".join(str(m) for m in raw_message)
elif isinstance(raw_message, str):
message_text = raw_message
else:
message_text = str(raw_message)
message = f"OpenWA API {status_code} — {context}: {message_text}"
return classify(status_code, message, body, envelope.get("error") if envelope else None)
class OpenWAAuthError(OpenWAApiError):
"""401 Unauthorized — missing or invalid API key."""
class OpenWAForbiddenError(OpenWAApiError):
"""403 Forbidden — insufficient role."""
class OpenWANotFoundError(OpenWAApiError):
"""404 Not Found."""
class OpenWAConflictError(OpenWAApiError):
"""409 Conflict — typically an engine-not-ready condition."""
class OpenWARateLimitError(OpenWAApiError):
"""429 Too Many Requests."""
class OpenWANotImplementedError(OpenWAApiError):
"""501 Not Implemented — the active engine does not support this operation."""
class OpenWATimeoutError(OpenWAError):
"""Raised when a request exceeds the configured timeout."""
def __init__(self, timeout: float) -> None:
super().__init__(f"Request timed out after {timeout}s")
self.timeout = timeout
def classify(status: int, message: str, body: Any, error_kind: str | None) -> OpenWAApiError:
"""Pick the most specific :class:`OpenWAApiError` subclass for a status."""
cls = {
401: OpenWAAuthError,
403: OpenWAForbiddenError,
404: OpenWANotFoundError,
409: OpenWAConflictError,
429: OpenWARateLimitError,
501: OpenWANotImplementedError,
}.get(status, OpenWAApiError)
return cls(message, status, body, error_kind)
|