FastapiAdmin/backend/app/core/exceptions.py

import traceback
from typing import Any

from fastapi import FastAPI, Request, status
from fastapi.exceptions import RequestValidationError, ResponseValidationError
from pydantic_validation_decorator import FieldValidationError
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse

from app.common.constant import RET
from app.common.response import ErrorResponse
from app.core.logger import logger


class CustomException(Exception):
    def __init__(
        self,
        msg: str = RET.EXCEPTION.msg,
        code: int = RET.EXCEPTION.code,
        status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR,
        data: Any | None = None,
        success: bool = False,
    ) -> None:
        super().__init__(msg)
        self.status_code = status_code
        self.code = code
        self.msg = msg
        self.data = data
        self.success = success

    def __str__(self) -> str:
        return self.msg


def _tb_source(exc: Exception) -> str:
    """从 traceback 提取抛出的 文件名:行号:函数名"""
    if exc.__traceback__ and (tb := traceback.extract_tb(exc.__traceback__)):
        f = tb[-1]
        return f"{f.filename}:{f.lineno}:{f.name}"
    return "unknown"


_VALIDATION_ERROR_MAP: dict[str, str] = {
    "Field required": "请求失败，缺少必填项！",
    "value is not a valid list": "类型错误，提交参数应该为列表！",
    "value is not a valid int": "类型错误，提交参数应该为整数！",
    "value could not be parsed to a boolean": "类型错误，提交参数应该为布尔值！",
    "Input should be a valid list": "类型错误，输入应该是一个有效的列表！",
}


def handle_exception(app: FastAPI) -> None:
    @app.exception_handler(CustomException)
    async def custom_exception_handler(request: Request, exc: CustomException) -> JSONResponse:
        logger.error(
            "[自定义异常] %s %s | source=%s | code=%s | msg=%s | data=%s",
            request.method, request.url.path,
            _tb_source(exc), exc.code, exc.msg, exc.data,
        )
        return ErrorResponse(msg=exc.msg, code=exc.code, status_code=exc.status_code, data=exc.data)

    @app.exception_handler(HTTPException)
    async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse:
        logger.error(
            "[HTTP异常] %s %s | status_code=%s | detail=%s",
            request.method, request.url.path, exc.status_code, exc.detail,
        )
        return ErrorResponse(msg=exc.detail, status_code=exc.status_code)

    @app.exception_handler(RequestValidationError)
    async def validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
        raw_msg = exc.errors()[0].get("msg")
        msg = _VALIDATION_ERROR_MAP.get(raw_msg, raw_msg)
        if isinstance(msg, str) and msg.startswith("Value error"):
            msg = msg[11:].lstrip(" ,")
        logger.error(
            "[参数验证异常] %s %s | msg=%s | errors=%s",
            request.method, request.url.path, msg, exc.errors(),
        )
        return ErrorResponse(msg=str(msg), status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, data=exc.body)

    @app.exception_handler(ResponseValidationError)
    async def response_validation_handler(request: Request, exc: ResponseValidationError) -> JSONResponse:
        logger.error(
            "[响应验证异常] %s %s | errors=%s",
            request.method, request.url.path, exc.errors(),
        )
        return ErrorResponse(msg="服务器响应格式错误", status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, data=exc.body)

    @app.exception_handler(SQLAlchemyError)
    async def sqlalchemy_exception_handler(request: Request, exc: SQLAlchemyError) -> JSONResponse:
        exc_type = type(exc).__name__
        logger.error(
            "[数据库异常] %s %s | type=%s | detail=%s",
            request.method, request.url.path, exc_type, exc,
        )

        if isinstance(exc, IntegrityError):
            detail = str(exc.orig) if exc.orig else str(exc)
            if "Duplicate entry" in detail:
                return ErrorResponse(msg="数据重复，请检查唯一字段", status_code=status.HTTP_409_CONFLICT, data=detail)
            if "foreign key constraint" in detail:
                return ErrorResponse(msg="存在关联数据，无法删除", status_code=status.HTTP_409_CONFLICT, data=detail)
            if "cannot be null" in detail:
                return ErrorResponse(msg="必填字段缺失", status_code=status.HTTP_409_CONFLICT, data=detail)
            return ErrorResponse(msg="数据已存在或违反完整性约束", status_code=status.HTTP_409_CONFLICT, data=detail)

        lower = str(exc).lower()
        if "connect" in lower or "connection" in lower:
            return ErrorResponse(msg="数据库连接失败", status_code=status.HTTP_503_SERVICE_UNAVAILABLE, data=exc_type)
        return ErrorResponse(msg=f"数据库操作失败: {exc_type}", status_code=status.HTTP_400_BAD_REQUEST, data=str(exc))

    @app.exception_handler(ValueError)
    async def value_exception_handler(request: Request, exc: ValueError) -> JSONResponse:
        logger.error("[值异常] %s %s | msg=%s", request.method, request.url.path, exc)
        return ErrorResponse(msg=str(exc), status_code=status.HTTP_400_BAD_REQUEST)

    @app.exception_handler(FieldValidationError)
    async def field_validation_exception_handler(request: Request, exc: FieldValidationError) -> JSONResponse:
        logger.error("[字段验证异常] %s %s | msg=%s", request.method, request.url.path, exc.message)
        return ErrorResponse(msg=exc.message, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)

    @app.exception_handler(Exception)
    async def all_exception_handler(request: Request, exc: Exception) -> JSONResponse:
        exc_type = type(exc).__name__
        logger.error(
            "[未捕获异常] %s %s | type=%s | detail=%s",
            request.method, request.url.path, exc_type, exc,
        )
        return ErrorResponse(msg="服务器内部错误", status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)