69 lines
2.2 KiB
Python
69 lines
2.2 KiB
Python
 
|
"""Every API error is JSON `{"detail": "<message>", "code": "<snake_code>"}`, with extra fields when useful."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from fastapi import FastAPI, Request
|
||
|
|
from fastapi.exceptions import RequestValidationError
|
||
|
|
from fastapi.responses import JSONResponse
|
||
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
||
|
|
|
||
|
|
HTTP_CODES = {
|
||
|
|
400: "bad_request",
|
||
|
|
401: "no_session",
|
||
|
|
403: "forbidden",
|
||
|
|
404: "not_found",
|
||
|
|
405: "method_not_allowed",
|
||
|
|
409: "conflict",
|
||
|
|
413: "too_large",
|
||
|
|
429: "rate_limited",
|
||
|
|
503: "unavailable",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
class ApiError(Exception):
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
status: int,
|
||
|
|
code: str,
|
||
|
|
detail: str,
|
||
|
|
*,
|
||
|
|
headers: dict[str, str] | None = None,
|
||
|
|
extra: dict[str, Any] | None = None,
|
||
|
|
):
|
||
|
|
super().__init__(detail)
|
||
|
|
self.status = status
|
||
|
|
self.code = code
|
||
|
|
self.detail = detail
|
||
|
|
self.headers = headers or {}
|
||
|
|
self.extra = extra or {}
|
||
|
|
|
||
|
|
def body(self) -> dict[str, Any]:
|
||
|
|
return {"detail": self.detail, "code": self.code, **self.extra}
|
||
|
|
|
||
|
|
|
||
|
|
def error_response(err: ApiError) -> JSONResponse:
|
||
|
|
return JSONResponse(err.body(), status_code=err.status, headers=err.headers)
|
||
|
|
|
||
|
|
|
||
|
|
def install(app: FastAPI) -> None:
|
||
|
|
@app.exception_handler(ApiError)
|
||
|
|
async def _api_error(_: Request, exc: ApiError) -> JSONResponse:
|
||
|
|
return error_response(exc)
|
||
|
|
|
||
|
|
@app.exception_handler(RequestValidationError)
|
||
|
|
async def _invalid(_: Request, exc: RequestValidationError) -> JSONResponse:
|
||
|
|
errors = [{"loc": [str(x) for x in e.get("loc", ())], "msg": str(e.get("msg", ""))} for e in exc.errors()]
|
||
|
|
where = ", ".join(".".join(e["loc"][1:]) or e["loc"][0] for e in errors if e["loc"]) or "request"
|
||
|
|
return JSONResponse(
|
||
|
|
{"detail": f"Invalid request: {where}", "code": "invalid_request", "errors": errors}, status_code=422
|
||
|
|
)
|
||
|
|
|
||
|
|
@app.exception_handler(StarletteHTTPException)
|
||
|
|
async def _http(_: Request, exc: StarletteHTTPException) -> JSONResponse:
|
||
|
|
code = HTTP_CODES.get(exc.status_code, "http_error")
|
||
|
|
return JSONResponse(
|
||
|
|
{"detail": str(exc.detail), "code": code}, status_code=exc.status_code, headers=exc.headers or None
|
||
|
|
)
|