Build a typed error from status + the parsed envelope fields.
(
*,
status: int,
code: str = "",
message: Optional[str] = None,
request_id: Optional[str] = None,
details: Any = None,
headers: Optional[Mapping[str, str]] = None,
cause: Optional[BaseException] = None,
)
| 245 | |
| 246 | |
| 247 | def to_e2a_error( |
| 248 | *, |
| 249 | status: int, |
| 250 | code: str = "", |
| 251 | message: Optional[str] = None, |
| 252 | request_id: Optional[str] = None, |
| 253 | details: Any = None, |
| 254 | headers: Optional[Mapping[str, str]] = None, |
| 255 | cause: Optional[BaseException] = None, |
| 256 | ) -> E2AError: |
| 257 | """Build a typed error from status + the parsed envelope fields.""" |
| 258 | resolved_code = code or _DEFAULT_CODE.get(status) or ( |
| 259 | "internal_error" if status >= 500 else "error" |
| 260 | ) |
| 261 | klass, retryable = _resolve(status, code) |
| 262 | # Prefer the Retry-After header; else fall back to the error body's |
| 263 | # details.retry_after_seconds. The send-path 429 carries its hint in the |
| 264 | # body (not the header) because of a Huma constraint, so we honor both. |
| 265 | retry_after = _parse_retry_after(headers) |
| 266 | if retry_after is None: |
| 267 | retry_after = _retry_after_from_details(details) |
| 268 | err = klass( |
| 269 | code=resolved_code, |
| 270 | message=message or f"e2a API error ({status})", |
| 271 | status=status, |
| 272 | request_id=request_id, |
| 273 | details=details, |
| 274 | retryable=retryable, |
| 275 | retry_after_seconds=retry_after, |
| 276 | ) |
| 277 | if cause is not None: |
| 278 | err.__cause__ = cause |
| 279 | return err |
| 280 | |
| 281 | |
| 282 | def from_api_exception(exc: ApiException) -> E2AError: |
no test coverage detected