PacketStation/server/packetstation/api/errors.py

"""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: Request, exc: ApiError) -> JSONResponse:
        principal = getattr(request.state, "principal", None)
        if exc.status == 403 and principal is not None:
            # A refusal to a signed-in person is audited as policy.denied: method, path and code, never the body
            from starlette.concurrency import run_in_threadpool

            from ..domain.security_alerts import record_denied

            await run_in_threadpool(
                record_denied,
                request.app.state.settings,
                principal,
                method=request.method,
                path=request.url.path,
                code=exc.code,
            )
        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
        )