Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/openstatusHQ/sdk-python. Official Python SDK for openstatus
Something went wrong. Try again.
2.1 kB · 73 lines
Python
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374from __future__ import annotations
import jsonfrom typing import NoReturn, cast
import httpx
from .exceptions import ( AuthenticationError, InvalidArgumentError, JsonValue, NotFoundError, OpenstatusError, PermissionDeniedError, RateLimitError, ServiceUnavailableError,)
_CODE_TO_EXCEPTION: dict[str, type[OpenstatusError]] = { "unauthenticated": AuthenticationError, "not_found": NotFoundError, "invalid_argument": InvalidArgumentError, "permission_denied": PermissionDeniedError, "resource_exhausted": RateLimitError, "unavailable": ServiceUnavailableError,}
class ConnectErrorMapper: """Decode a Connect RPC error response and raise the matching SDK exception."""
@staticmethod def rethrow(response: httpx.Response) -> NoReturn: raw = response.text http_status = response.status_code fallback_message = response.reason_phrase or f"HTTP {http_status}"
envelope = _safe_load_envelope(raw) if envelope is None: raise ServiceUnavailableError( fallback_message, connect_code="unknown", http_status=http_status, raw_body=raw, )
code_raw = envelope.get("code") code = code_raw if isinstance(code_raw, str) else "unknown"
message_raw = envelope.get("message") message = message_raw if isinstance(message_raw, str) else fallback_message
details_raw = envelope.get("details") details: list[JsonValue] = details_raw if isinstance(details_raw, list) else []
exc_cls = _CODE_TO_EXCEPTION.get(code, OpenstatusError) raise exc_cls( message, connect_code=code, http_status=http_status, details=details, raw_body=raw, )
def _safe_load_envelope(raw: str) -> dict[str, JsonValue] | None: try: parsed: object = json.loads(raw) except json.JSONDecodeError: return None if not isinstance(parsed, dict): return None return cast(dict[str, JsonValue], parsed)