File size: 3,088 Bytes
7302343 | 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 | """Error envelope + exception handlers.
Wire format per docs/Specs.md §10.3:
{ "ok": false, "error": { "code": "...", "message": "...", "retryable": bool } }
"""
from typing import Literal
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from starlette.exceptions import HTTPException as StarletteHTTPException
ErrorCode = Literal[
"BAD_REQUEST",
"UNAUTHORIZED",
"RATE_LIMITED",
"BUDGET_EXHAUSTED",
"ENGINE_DEGRADED",
"TIMEOUT",
"INTERNAL",
]
_RETRYABLE: set[ErrorCode] = {"RATE_LIMITED", "ENGINE_DEGRADED", "TIMEOUT"}
_STATUS_FOR_CODE: dict[ErrorCode, int] = {
"BAD_REQUEST": 400,
"UNAUTHORIZED": 401,
"RATE_LIMITED": 429,
"BUDGET_EXHAUSTED": 429,
"ENGINE_DEGRADED": 503,
"TIMEOUT": 504,
"INTERNAL": 500,
}
class ErrorBody(BaseModel):
code: ErrorCode
message: str
retryable: bool
class ErrorEnvelope(BaseModel):
ok: Literal[False] = False
error: ErrorBody
class EngineError(Exception):
"""Raised by handlers; serialized to the canonical error envelope."""
def __init__(self, code: ErrorCode, message: str) -> None:
super().__init__(message)
self.code = code
self.message = message
@property
def retryable(self) -> bool:
return self.code in _RETRYABLE
@property
def status_code(self) -> int:
return _STATUS_FOR_CODE[self.code]
def _envelope(code: ErrorCode, message: str) -> JSONResponse:
body = ErrorEnvelope(error=ErrorBody(code=code, message=message, retryable=code in _RETRYABLE))
return JSONResponse(status_code=_STATUS_FOR_CODE[code], content=body.model_dump())
def register_error_handlers(app: FastAPI) -> None:
@app.exception_handler(EngineError)
async def _engine_error(_req: Request, exc: EngineError) -> JSONResponse:
return _envelope(exc.code, exc.message)
@app.exception_handler(RequestValidationError)
async def _validation_error(_req: Request, exc: RequestValidationError) -> JSONResponse:
# Pydantic / FastAPI validation failure -> BAD_REQUEST envelope
message = "; ".join(
f"{'.'.join(str(p) for p in e['loc'])}: {e['msg']}" for e in exc.errors()
)
return _envelope("BAD_REQUEST", message or "validation failed")
@app.exception_handler(StarletteHTTPException)
async def _http_error(_req: Request, exc: StarletteHTTPException) -> JSONResponse:
code: ErrorCode = (
"UNAUTHORIZED" if exc.status_code == 401
else "BAD_REQUEST" if exc.status_code in (400, 404, 405)
else "RATE_LIMITED" if exc.status_code == 429
else "INTERNAL"
)
# Preserve the original HTTP status — `_envelope` would re-map it via _STATUS_FOR_CODE.
body = ErrorEnvelope(
error=ErrorBody(code=code, message=str(exc.detail), retryable=code in _RETRYABLE)
)
return JSONResponse(status_code=exc.status_code, content=body.model_dump())
|