File size: 2,577 Bytes
0fd3d47
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Exception handlers.

The previous global handler returned ``str(exc)`` in the response body. For a
SQLAlchemy failure that string contains the connection URL, including the password,
so a single database hiccup leaked the credentials to whoever triggered it.

Here the traceback is logged server-side against a request id, and the client gets
only that id. In development the detail is included, because debugging locally
against an opaque error is miserable and there is no secret worth protecting.
"""

from __future__ import annotations

import logging

from fastapi import FastAPI, HTTPException, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

from app.core.config import Settings

logger = logging.getLogger("facevision.errors")


def _request_id(request: Request) -> str:
    return getattr(request.state, "request_id", "unknown")


def register_exception_handlers(app: FastAPI, settings: Settings) -> None:
    """Attach the handlers to ``app``."""

    @app.exception_handler(RequestValidationError)
    async def validation_exception_handler(
        request: Request, exc: RequestValidationError
    ) -> JSONResponse:
        # Pydantic's own error structures are safe to return: they describe the
        # caller's input, not the server's internals.
        return JSONResponse(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            content={
                "message": "Validation failed",
                "request_id": _request_id(request),
                "errors": exc.errors(),
            },
        )

    @app.exception_handler(HTTPException)
    async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse:
        return JSONResponse(
            status_code=exc.status_code,
            content={"message": exc.detail, "request_id": _request_id(request)},
            headers=exc.headers,
        )

    @app.exception_handler(Exception)
    async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
        request_id = _request_id(request)
        logger.error(
            "unhandled exception",
            exc_info=exc,
            extra={"request_id": request_id, "path": request.url.path},
        )

        content: dict[str, object] = {
            "message": "Internal Server Error",
            "request_id": request_id,
        }
        if settings.is_development:
            content["detail"] = f"{type(exc).__name__}: {exc}"

        return JSONResponse(status_code=500, content=content)