异常处理

前言

异常处理是 API 开发的必修课：业务逻辑抛异常、参数校验失败、未捕获的异常，都需要以统一的响应格式返回给客户端。FastAPI 的异常处理分两层：

• 抛出异常：业务代码中 raise HTTPException(...) 或自定义异常。
• 全局拦截：注册 exception_handler，统一加工后再返回。

HTTPException

HTTPException 用于在业务逻辑中抛出指定状态码的 HTTP 异常，FastAPI 会将其转换为对应响应返回给客户端。

from fastapi import HTTPException, Query


@router.get("/test1", summary="异常处理测试1")
async def testexcept1(action_scopes: str = Query(default="admin")):
    if action_scopes == "admin":
        raise HTTPException(
            status_code=403,
            headers={
                "x-auth": "NO AUTH",
            },
            detail={
                "code": 403,
                "message": "permission denied",
            },
        )
    return {"code": 200}

说明：

• detail 可以是任意可 JSON 序列化的值（字符串、dict、list 等）。
• headers 用于在错误响应中附带自定义请求头（如 WWW-Authenticate）。
• 注意 HTTPException 是抛出（raise）而不是返回（return）。

全局拦截异常

只抛异常还不够，很多场景需要统一拦截、加工后再返回，可以通过全局异常处理器实现。

方式一：装饰器注册（官方推荐）

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

app = FastAPI()


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


@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"message": "Internal Server Error"},
    )

方式二：构造器注册（全局集中管理）

import traceback
from http import HTTPStatus
from typing import Any, Mapping

import uvicorn
from fastapi import FastAPI, Request
from fastapi.exceptions import HTTPException
from fastapi.responses import JSONResponse, Response


async def exception_general(request: Request, exc: Exception) -> Response:
    print(f"Exception occurred: {traceback.format_exc()}")
    return JSONResponse(
        status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
        content={"message": "Internal Server Error"},
    )


async def exception_not_found(request: Request, exc: HTTPException) -> Response:
    return JSONResponse(
        status_code=HTTPStatus.NOT_FOUND,
        content={"message": "Not Found", "detail": exc.detail},
        headers=exc.headers,
    )


async def exception_not_auth(request: Request, exc: HTTPException) -> Response:
    return JSONResponse(
        status_code=HTTPStatus.UNAUTHORIZED,
        content={"message": "Unauthorized", "detail": exc.detail},
        headers=exc.headers,
    )


exception_handlers: Mapping[Any, Any] = {
    Exception: exception_general,
    HTTPStatus.NOT_FOUND: exception_not_found,
    HTTPStatus.UNAUTHORIZED: exception_not_auth,
}

app = FastAPI(exception_handlers=exception_handlers)


@app.get("/q")
async def greeting(name: str):
    if name == "qqq":
        raise Exception("General exception triggered")
    return {"message": f"Hello {name}"}

两种注册方式对比：

方式	写法	适用场景
装饰器	@app.exception_handler(XxxException)	官方文档推荐，随用随注册
构造器	FastAPI(exception_handlers={...})	全局异常集中管理

RequestValidationError

对请求体、表单、路径参数、查询参数等参数进行校验时，如果校验失败，FastAPI 会抛出 RequestValidationError，默认返回 422 响应。

比如下面的路由要求查询参数 user_id 必须是整型：

@router.get("/reqvalidtest1", summary="校验测试")
async def reqvalidtest1(user_id: int):
    return {"user_id": user_id}

请求测试：

curl -s 'http://127.0.0.1:8000/exp/reqvalidtest1?user_id=qwer' | python3 -m json.tool

结果：

{
    "detail": [
        {
            "type": "int_parsing",
            "loc": [
                "query",
                "user_id"
            ],
            "msg": "Input should be a valid integer, unable to parse string as an integer",
            "input": "qwer",
            "url": "https://errors.pydantic.dev/2.5/v/int_parsing"
        }
    ]
}

全局拦截 RequestValidationError

from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()


@app.exception_handler(RequestValidationError)
async def exception_valid(request, exc):
    return JSONResponse(
        status_code=422,
        content={
            "code": 422,
            "msg": "Invalid Arguments",
            # 建议用 exc.errors() 逐条取错误，而不是直接 str(exc)
            "data": exc.errors(),
        },
    )

备注

关于两种注册方式的结论（实测于 FastAPI 0.141.1）：构造器注册 FastAPI(exception_handlers={RequestValidationError: handler}) 也是可行的。旧版本（约 0.100 之前）FastAPI 在 setup() 中先注册用户传入的 handler、再注册默认 handler，导致用户自定义的 RequestValidationError handler 被默认 handler 覆盖；新版本改用 setdefault 保证用户注册优先，两种方式行为一致。官方文档示例使用装饰器，推荐优先使用装饰器；如果想集中管理所有全局异常，也可以使用构造器。

注意事项

• RequestValidationError 包含校验错误的文件、行号等信息，如果直接 str(exc) 返回给客户端，可能泄露系统内部信息，建议用 exc.errors() 逐条提取后返回。
• 需要复用 FastAPI 默认行为时，可以引入 fastapi.exception_handlers.request_validation_exception_handler 在自定义 handler 中调用。

自定义异常

在 FastAPI 中，错误和异常都继承自 Exception，所以自定义异常可以继承 Exception 或者其它已实现的 Exception 子类。

基本实现

class CustomException(Exception):
    def __init__(self, message: str):
        self.message = message


# 全局拦截
@app.exception_handler(CustomException)
async def exception_custom(request, exc):
    return JSONResponse(
        content={
            "msg": exc.message,
        }
    )


# 测试路由
@app.get("/custom", summary="自定义异常")
async def customtest(name: str = 'zhangsan'):
    if name == 'zhangsan':
        raise CustomException("自定义异常")
    return {"name": name}

自定义内部错误码和异常

通过自定义错误码，开发人员可以对错误进行快速定位。响应格式示例：

{
    "return_code": "SUCCESS", # SUCCESS/FAIL 是通信标识，非业务标识
    "return_msg": "OK", # 当return_code为FAIL时返回错误原因
    "err_code": "SYSTEMERROR", # 错误代码
    "err_code_des": "系统错误" # 错误描述
}

先枚举错误码区间：

from enum import Enum


class ExceptionEnum(Enum):
    SUCCESS = ("0000", "OK")
    FAIL = ("9999", "FAIL")

    USER_NO_DATA = ("1001", "User Not Exist")
    USER_REGIST_FAIL = ("1002", "User Register Fail")
    USER_LOGIN_FAIL = ("1003", "User Login Fail")
    USER_PERMISSION_FAIL = ("1004", "User Permission Fail")

根据错误自定义异常类：

class BusinessError(Exception):
    # __slots__ 用于声明实例可以拥有的属性列表，限制动态属性，节省内存
    __slots__ = ["err_code", "err_code_des"]

    def __init__(
        self,
        result: ExceptionEnum | None = None,
        err_code: str = "00000",
        err_code_des: str = "",
    ):
        if result:
            self.err_code = result.value[0]
            self.err_code_des = result.value[1]
        else:
            self.err_code = err_code
            self.err_code_des = err_code_des
        super().__init__(f"{self.err_code}: {self.err_code_des}")

添加全局错误拦截：

@app.exception_handler(BusinessError)
async def exception_business(request, exc):
    return JSONResponse(
        content={
            "return_code": "FAIL",
            "return_msg": "Invalid Arguments",
            "err_code": exc.err_code,
            "err_code_des": exc.err_code_des,
        }
    )

添加路由测试：

@app.get("/custom2", summary="自定义异常")
async def customtest2(name: str = 'zhangsan'):
    if name == 'zhangsan':
        raise BusinessError(ExceptionEnum.USER_LOGIN_FAIL)
    return {"name": name}

测试响应输出：

{
  "return_code": "FAIL",
  "return_msg": "Invalid Arguments",
  "err_code": "1003",
  "err_code_des": "User Login Fail"
}

中间件异常与全局处理

在 FastAPI 中间件中抛出的异常，无法被注册的自定义异常处理器捕获。原因是：FastAPI 底层把所有中间件抛出的异常统一交给顶层的 ServerErrorMiddleware 处理，而该中间件只按 Exception 处理，最终返回 500。

因此，中间件内的异常建议直接处理并返回对应响应，而不是抛出后指望全局处理器兜底：

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse


class ExceptionHandleMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            return await call_next(request)
        except Exception as exc:  # noqa: BLE001
            # 中间件内自行处理异常，返回统一的错误响应
            return JSONResponse(
                status_code=500,
                content={"code": 500, "msg": "Internal Server Error", "detail": str(exc)},
            )


app.add_middleware(ExceptionHandleMiddleware)

提示

业务逻辑（路由函数）里的异常可以正常使用全局异常处理器；只有中间件自身抛出的异常需要这样处理。更多中间件用法见 中间件。