- 微信
- 微博
  
  分享文章到微博
- 复制链接
  
  复制链接到剪贴板

FastAPI 统一异常处理与 API 响应规范实战：自定义异常、全局处理器与统一返回结构

yd_228838069 发表于 2026/11/06 08:52:56 2026/11/06

【摘要】在实际开发中，一个 Web 项目通常会包含大量 API 接口。例如：用户注册、登录用户信息管理文章发布与编辑文件上传与下载数据查询与分页权限验证后台管理随着接口数量增加，如果每个接口都单独编写异常处理和返回结构，就容易出现各种不一致的问题。例如，同样是用户不存在，不同接口可能返回：{ "detail": "用户不存在"}另一个接口却返回：{ "code": 404, "message"...

在实际开发中，一个 Web 项目通常会包含大量 API 接口。

例如：

用户注册、登录
用户信息管理
文章发布与编辑
文件上传与下载
数据查询与分页
权限验证
后台管理

随着接口数量增加，如果每个接口都单独编写异常处理和返回结构，就容易出现各种不一致的问题。

例如，同样是用户不存在，不同接口可能返回：

{
  "detail": "用户不存在"
}

另一个接口却返回：

{
  "code": 404,
  "message": "User Not Found"
}

还有一些接口可能直接返回数据库异常或 Python 错误信息。

这会给前端开发、接口维护和问题排查带来麻烦。

本文将使用 FastAPI 从零实现一套统一异常处理与 API 响应规范，包括：

统一成功响应结构。
统一错误响应结构。
自定义业务异常。
全局异常处理器。
HTTP 异常处理。
参数验证异常处理。
数据库异常处理。
请求日志与异常追踪。
Vue 3 前端统一处理 API 响应。
生产环境异常信息保护。

一、为什么需要统一 API 响应规范？

假设一个系统有 100 个 API 接口。

如果每个接口都采用不同的返回结构，前端就必须针对不同接口编写不同的判断逻辑。

例如：

if (response.data.code === 200) {
  // 处理成功
}

另一个接口可能是：

if (response.data.status === 'success') {
  // 处理成功
}

还有一个接口直接返回：

if (response.data.id) {
  // 处理成功
}

这样的代码很难维护。

因此，建议在项目中制定统一的 API 响应规范。

1. 统一成功响应

建议采用以下结构：

{
  "code": 200,
  "message": "操作成功",
  "data": {
    "id": 1,
    "username": "alice"
  }
}

字段说明：

字段	说明
code	业务状态码
message	提示信息
data	返回的数据

2. 统一错误响应

例如用户不存在：

{
  "code": 4004,
  "message": "用户不存在",
  "data": null
}

字段说明：

code：业务错误码。
message：面向调用方的错误提示。
data：错误时通常为 null。

这里需要区分两个概念：

HTTP 状态码用于表达 HTTP 请求的处理结果，例如 200、400、401、403、404、500。

业务状态码用于表达应用内部定义的业务结果，例如用户不存在、余额不足、重复提交等。

两者不应该混为一谈。

例如：

场景	HTTP 状态码	业务状态码
请求成功	200	2000
参数错误	422	4001
用户不存在	404	4004
未登录	401	4001
权限不足	403	4003
服务器内部错误	500	5000

本文后面会采用这套结构。

二、创建项目结构

创建项目：

mkdir fastapi-response-demo
cd fastapi-response-demo

创建虚拟环境：

python3 -m venv venv

Linux 激活环境：

source venv/bin/activate

Windows 激活环境：

venv\Scripts\activate

安装依赖：

pip install fastapi uvicorn

项目结构：

fastapi-response-demo/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── response.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── exceptions.py
│   │   └── handlers.py
│   └── routers/
│       ├── __init__.py
│       └── users.py
└── requirements.txt

各目录作用：

schemas：定义统一响应模型。
core/exceptions.py：定义自定义业务异常。
core/handlers.py：注册全局异常处理器。
routers：业务 API 接口。
main.py：应用入口。

三、定义统一响应模型

首先创建 app/schemas/response.py：

from typing import Generic, TypeVar

from pydantic import BaseModel, ConfigDict

T = TypeVar("T")


class APIResponse(BaseModel, Generic[T]):
    """
    统一 API 响应结构
    """

    code: int = 2000

    message: str = "操作成功"

    data: T | None = None

    model_config = ConfigDict(
        json_schema_extra={
            "examples": [
                {
                    "code": 2000,
                    "message": "操作成功",
                    "data": {
                        "id": 1,
                        "username": "alice"
                    }
                }
            ]
        }
    )

这里使用 Pydantic 泛型模型。

T 表示响应数据的类型。

例如：

APIResponse[dict]

表示 data 是字典。

APIResponse[list]

表示 data 是列表。

这样可以让不同接口使用相同的响应结构，同时保留类型提示。

四、编写统一成功响应函数

创建 app/core/responses.py：

from typing import Any


def success_response(
    data: Any = None,
    message: str = "操作成功",
    code: int = 2000
):
    """
    统一成功响应
    """

    return {
        "code": code,
        "message": message,
        "data": data
    }

例如：

from app.core.responses import success_response


@router.get("/users")
async def get_users():
    users = [
        {
            "id": 1,
            "username": "alice"
        }
    ]

    return success_response(
        data=users,
        message="获取用户列表成功"
    )

响应结果：

{
  "code": 2000,
  "message": "获取用户列表成功",
  "data": [
    {
      "id": 1,
      "username": "alice"
    }
  ]
}

需要注意，直接返回字典时，FastAPI 会自动将其转换为 JSON 响应。

如果希望使用 Pydantic 模型进行响应校验，也可以在路由中声明 response_model。

五、定义自定义业务异常

在实际业务中，很多错误并不属于 Python 运行错误，而是业务规则不允许。

例如：

用户不存在。
用户名已经注册。
文章已经删除。
账户余额不足。
当前用户没有操作权限。
文件已经被删除。

这些情况可以通过自定义异常统一管理。

创建 app/core/exceptions.py：

class BusinessException(Exception):
    """
    业务异常基类
    """

    def __init__(
        self,
        message: str = "业务处理失败",
        code: int = 4000,
        status_code: int = 400
    ):
        self.message = message
        self.code = code
        self.status_code = status_code

        super().__init__(message)


class UserNotFoundException(BusinessException):
    """
    用户不存在
    """

    def __init__(self):
        super().__init__(
            message="用户不存在",
            code=4004,
            status_code=404
        )


class UsernameExistsException(BusinessException):
    """
    用户名已经存在
    """

    def __init__(self):
        super().__init__(
            message="用户名已经存在",
            code=4005,
            status_code=409
        )


class PermissionDeniedException(BusinessException):
    """
    权限不足
    """

    def __init__(self):
        super().__init__(
            message="没有权限执行此操作",
            code=4003,
            status_code=403
        )

这样，业务代码就不需要反复编写相同的错误返回结构。

例如：

raise UserNotFoundException()

而不是在每个接口中手动返回错误 JSON。

六、实现全局异常处理器

这是本文最重要的部分。

FastAPI 支持通过 @app.exception_handler() 注册全局异常处理器。

创建 app/core/handlers.py：

import logging

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.responses import JSONResponse

from app.core.exceptions import BusinessException

logger = logging.getLogger(__name__)


async def business_exception_handler(
    request: Request,
    exc: BusinessException
):
    """
    业务异常处理
    """

    return JSONResponse(
        status_code=exc.status_code,
        content={
            "code": exc.code,
            "message": exc.message,
            "data": None
        }
    )


async def http_exception_handler(
    request: Request,
    exc: StarletteHTTPException
):
    """
    HTTP 异常处理
    """

    status_messages = {
        400: "请求参数错误",
        401: "请先登录",
        403: "没有权限",
        404: "请求的资源不存在",
        405: "请求方法不允许",
        409: "资源冲突",
        429: "请求过于频繁",
    }

    message = status_messages.get(
        exc.status_code,
        "请求处理失败"
    )

    # 如果业务主动提供了 detail，则保留其内容。
    if isinstance(exc.detail, str):
        message = exc.detail

    return JSONResponse(
        status_code=exc.status_code,
        content={
            "code": exc.status_code * 10,
            "message": message,
            "data": None
        },
        headers=exc.headers
    )


async def validation_exception_handler(
    request: Request,
    exc: RequestValidationError
):
    """
    请求参数验证异常
    """

    errors = []

    for error in exc.errors():
        location = ".".join(
            str(item)
            for item in error["loc"]
        )

        errors.append({
            "field": location,
            "message": error["msg"],
            "type": error["type"]
        })

    return JSONResponse(
        status_code=422,
        content={
            "code": 4001,
            "message": "请求参数验证失败",
            "data": {
                "errors": errors
            }
        }
    )


async def unhandled_exception_handler(
    request: Request,
    exc: Exception
):
    """
    未处理异常
    """

    logger.exception(
        "服务器内部异常: %s %s",
        request.method,
        request.url.path
    )

    return JSONResponse(
        status_code=500,
        content={
            "code": 5000,
            "message": "服务器内部错误",
            "data": None
        }
    )


def register_exception_handlers(app: FastAPI):
    """
    注册所有全局异常处理器
    """

    app.add_exception_handler(
        BusinessException,
        business_exception_handler
    )

    app.add_exception_handler(
        StarletteHTTPException,
        http_exception_handler
    )

    app.add_exception_handler(
        RequestValidationError,
        validation_exception_handler
    )

    app.add_exception_handler(
        Exception,
        unhandled_exception_handler
    )

这里注册了四种处理器。

1. 业务异常处理器

负责处理自己定义的业务异常。

例如：

raise UserNotFoundException()

最终会返回：

{
  "code": 4004,
  "message": "用户不存在",
  "data": null
}

HTTP 状态码为 404。

2. HTTP 异常处理器

负责处理 FastAPI 和 Starlette 中的 HTTP 异常。

例如：

raise HTTPException(
    status_code=403,
    detail="禁止访问"
)

最终会返回统一结构。

同时保留原始异常中的 HTTP 响应头，例如认证失败时的 WWW-Authenticate。

3. 参数验证异常处理器

FastAPI 使用 Pydantic 验证请求参数。

例如：

from pydantic import BaseModel, Field


class UserCreate(BaseModel):
    username: str = Field(min_length=3)
    age: int = Field(ge=1, le=120)

如果用户提交：

{
  "username": "ab",
  "age": -1
}

FastAPI 会自动触发 RequestValidationError。

经过全局处理后，返回：

{
  "code": 4001,
  "message": "请求参数验证失败",
  "data": {
    "errors": [
      {
        "field": "body.username",
        "message": "String should have at least 3 characters",
        "type": "string_too_short"
      },
      {
        "field": "body.age",
        "message": "Input should be greater than or equal to 1",
        "type": "greater_than_equal"
      }
    ]
  }
}

这样前端可以根据 field 显示具体字段的验证错误。

注意，参数错误信息可能包含框架内部细节。生产环境可以将底层错误消息映射成经过整理的用户提示。

4. 未处理异常处理器

例如代码中出现：

result = 1 / 0

会触发 Python 的 ZeroDivisionError。

全局异常处理器会：

将完整异常堆栈写入服务端日志。
向客户端返回统一的 500 响应。
不向客户端暴露 Python 文件路径、数据库连接信息和堆栈内容。

这对生产环境尤其重要。

七、在 main.py 中注册异常处理器

修改 app/main.py：

from fastapi import FastAPI

from app.core.handlers import register_exception_handlers
from app.routers.users import router as users_router

app = FastAPI(
    title="FastAPI 统一响应示例",
    version="1.0.0"
)

# 注册全局异常处理器
register_exception_handlers(app)

# 注册路由
app.include_router(users_router)


@app.get("/")
async def index():
    return {
        "message": "服务运行正常"
    }

现在整个 FastAPI 应用都可以使用统一异常处理机制。

八、编写用户管理接口

接下来创建 app/routers/users.py：

from fastapi import APIRouter, HTTPException

from pydantic import BaseModel, Field

from app.core.responses import success_response

from app.core.exceptions import (
    UserNotFoundException,
    UsernameExistsException,
    PermissionDeniedException,
)

router = APIRouter(
    prefix="/api/users",
    tags=["用户管理"]
)


# 教学示例：内存数据
users = [
    {
        "id": 1,
        "username": "alice",
        "age": 25
    },
    {
        "id": 2,
        "username": "bob",
        "age": 30
    }
]


class UserCreate(BaseModel):
    username: str = Field(
        min_length=3,
        max_length=30
    )

    age: int = Field(
        ge=1,
        le=120
    )


@router.get("")
async def get_users():
    return success_response(
        data=users,
        message="获取用户列表成功"
    )


@router.get("/{user_id}")
async def get_user(user_id: int):

    user = next(
        (
            item
            for item in users
            if item["id"] == user_id
        ),
        None
    )

    if not user:
        raise UserNotFoundException()

    return success_response(
        data=user,
        message="获取用户信息成功"
    )


@router.post("")
async def create_user(user: UserCreate):

    # 检查用户名是否重复
    exists = any(
        item["username"] == user.username
        for item in users
    )

    if exists:
        raise UsernameExistsException()

    new_user = {
        "id": max(
            (item["id"] for item in users),
            default=0
        ) + 1,
        "username": user.username,
        "age": user.age
    }

    users.append(new_user)

    return success_response(
        data=new_user,
        message="用户创建成功"
    )


@router.delete("/{user_id}")
async def delete_user(user_id: int):

    user = next(
        (
            item
            for item in users
            if item["id"] == user_id
        ),
        None
    )

    if not user:
        raise UserNotFoundException()

    users.remove(user)

    return success_response(
        message="用户删除成功"
    )


@router.get("/test/forbidden")
async def test_forbidden():
    raise PermissionDeniedException()


@router.get("/test/http-error")
async def test_http_error():
    raise HTTPException(
        status_code=403,
        detail="当前账号禁止访问"
    )


@router.get("/test/server-error")
async def test_server_error():
    return 1 / 0

这里使用内存列表模拟数据库，只用于演示异常处理流程。

正式项目应将用户查询、创建和删除替换为数据库操作。

九、启动项目并测试接口

启动服务：

uvicorn app.main:app --reload

打开：

http://127.0.0.1:8000/docs

依次测试以下接口。

1. 测试成功响应

请求：

GET /api/users

{
  "code": 2000,
  "message": "获取用户列表成功",
  "data": [
    {
      "id": 1,
      "username": "alice",
      "age": 25
    },
    {
      "id": 2,
      "username": "bob",
      "age": 30
    }
  ]
}

2. 测试业务异常

请求：

GET /api/users/999

返回 HTTP 404：

{
  "code": 4004,
  "message": "用户不存在",
  "data": null
}

3. 测试用户名重复

请求：

POST /api/users

请求体：

{
  "username": "alice",
  "age": 25
}

返回 HTTP 409：

{
  "code": 4005,
  "message": "用户名已经存在",
  "data": null
}

4. 测试参数验证

请求：

POST /api/users

请求体：

{
  "username": "ab",
  "age": -1
}

返回 HTTP 422，并包含字段错误信息。

5. 测试权限异常

请求：

GET /api/users/test/forbidden

返回 HTTP 403：

{
  "code": 4003,
  "message": "没有权限执行此操作",
  "data": null
}

6. 测试服务器异常

请求：

GET /api/users/test/server-error

返回 HTTP 500：

{
  "code": 5000,
  "message": "服务器内部错误",
  "data": null
}

服务端日志中则会保留异常堆栈，便于开发人员排查。

十、Vue 3 前端统一处理 API 响应

后端返回结构统一后，前端也应该统一处理。

以 Vue 3 + Axios 为例。

安装 Axios：

npm install axios

创建 src/utils/request.js：

import axios from 'axios'

const request = axios.create({
  baseURL: '/api',
  timeout: 15000
})

// 请求拦截器
request.interceptors.request.use(
  config => {
    const token = localStorage.getItem('access_token')

    if (token) {
      config.headers.Authorization = `Bearer ${token}`
    }

    return config
  },
  error => Promise.reject(error)
)

// 响应拦截器
request.interceptors.response.use(
  response => {
    const result = response.data

    if (result.code !== 2000) {
      console.error(result.message || '请求失败')

      return Promise.reject(
        new Error(result.message || '请求失败')
      )
    }

    return result
  },

  error => {
    const status = error.response?.status

    if (status === 401) {
      console.error('登录状态已失效')
    } else if (status === 403) {
      console.error('没有操作权限')
    } else if (status === 404) {
      console.error('请求的资源不存在')
    } else if (status >= 500) {
      console.error('服务器发生错误')
    } else {
      console.error(
        error.response?.data?.message || '网络请求失败'
      )
    }

    return Promise.reject(error)
  }
)

export default request

这里假设所有业务接口成功时都返回 code: 2000。

1. 调用用户列表接口

创建 src/api/users.js：

import request from '@/utils/request'

export function getUsers() {
  return request.get('/users')
}

export function getUser(id) {
  return request.get(`/users/${id}`)
}

export function createUser(data) {
  return request.post('/users', data)
}

export function deleteUser(id) {
  return request.delete(`/users/${id}`)
}

2. 在 Vue 页面中使用

<script setup>
import { ref, onMounted } from 'vue'
import { getUsers } from '@/api/users'

const users = ref([])
const loading = ref(false)

async function loadUsers() {
  loading.value = true

  try {
    const response = await getUsers()

    users.value = response.data

  } catch (error) {
    console.error('获取用户列表失败', error)

  } finally {
    loading.value = false
  }
}

onMounted(() => {
  loadUsers()
})
</script>

因为 Axios 响应拦截器已经返回了 response.data，所以这里的 response.data 就是后端统一响应中的业务数据。

这样，页面不需要每次都判断 code。

十一、请求日志与异常追踪

统一异常处理只是第一步。

在生产环境中，还需要知道：

哪个接口发生了异常？
请求使用了什么 HTTP 方法？
哪个请求耗时较长？
异常发生在什么时间？
如何将一次请求与日志关联起来？

可以通过中间件记录请求信息。

创建 app/core/middleware.py：

import logging
import time
import uuid

from starlette.middleware.base import BaseHTTPMiddleware

logger = logging.getLogger("api")


class RequestLogMiddleware(BaseHTTPMiddleware):

    async def dispatch(self, request, call_next):

        request_id = uuid.uuid4().hex

        request.state.request_id = request_id

        start_time = time.perf_counter()

        try:
            response = await call_next(request)

        except Exception:
            logger.exception(
                "请求处理异常 request_id=%s method=%s path=%s",
                request_id,
                request.method,
                request.url.path
            )
            raise

        duration = time.perf_counter() - start_time

        response.headers["X-Request-ID"] = request_id

        logger.info(
            "request_id=%s method=%s path=%s status=%s duration=%.3f",
            request_id,
            request.method,
            request.url.path,
            response.status_code,
            duration
        )

        return response

在 main.py 中注册：

from app.core.middleware import RequestLogMiddleware

app.add_middleware(RequestLogMiddleware)

现在每次请求都会记录：

请求 ID
请求方法
请求路径
HTTP 状态码
请求耗时

需要注意，Starlette 的中间件与异常处理器存在调用层级关系。未处理异常可能在中间件外层重新抛出，因此中间件日志和全局异常日志可能同时记录同一个异常。生产环境应统一日志字段和请求 ID，避免重复告警。

另外，不要在日志中直接记录密码、JWT、完整 Cookie、银行卡号等敏感信息。

十二、生产环境需要注意的几个问题

1. 不要把所有异常都返回 HTTP 200

有些项目会采用：

{
  "code": 5000,
  "message": "服务器错误",
  "data": null
}

但 HTTP 状态码仍然是 200。

这并不是绝对不可行，但会让浏览器、代理服务器、监控系统和客户端缓存难以正确识别请求失败。

更推荐：

成功返回 HTTP 200、201 或 204。
参数错误返回 HTTP 400 或 422。
未登录返回 HTTP 401。
权限不足返回 HTTP 403。
资源不存在返回 HTTP 404。
服务器异常返回 HTTP 500。

业务状态码则用于进一步区分具体业务情况。

2. 不要直接向前端返回异常堆栈

错误示例：

except Exception as exc:
    return {
        "error": str(exc)
    }

这样可能暴露：

数据库连接地址。
服务器文件路径。
SQL 语句。
内部模块结构。
第三方服务信息。

正确做法是：

服务端记录完整异常，客户端只返回经过处理的错误提示。

3. 不要滥用业务异常

业务异常适合表达可预期的业务失败。

例如：

raise UserNotFoundException()

不应该用来代替所有 Python 错误。

例如数据库连接失败、代码逻辑错误等，应该由对应的异常处理和日志系统处理。

4. 不要把所有响应都强制包装

有些接口不一定适合统一 JSON 包装。

例如：

文件下载。
图片访问。
视频流。
SSE 实时事件。
WebSocket。
HTTP 204 无内容响应。

这些接口应该遵循相应的 HTTP 协议语义，不要强行套用：

{
  "code": 2000,
  "message": "成功",
  "data": null
}

统一响应规范主要针对普通 JSON API。

5. 区分开发环境与生产环境

开发环境可以提供更多错误信息。

生产环境应当隐藏内部实现细节。

例如：

DEBUG = False

并且应该关闭不必要的调试信息，妥善管理日志和异常追踪系统。

十三、最终项目结构

完成本文后，项目结构如下：

fastapi-response-demo/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── response.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── responses.py
│   │   ├── exceptions.py
│   │   ├── handlers.py
│   │   └── middleware.py
│   └── routers/
│       ├── __init__.py
│       └── users.py
└── requirements.txt

requirements.txt：

fastapi
uvicorn

安装依赖：

pip install -r requirements.txt

启动服务：

uvicorn app.main:app --host 0.0.0.0 --port 8000

十四、总结

本文完成了一套 FastAPI 统一异常处理与 API 响应规范。

主要内容包括：

使用 Pydantic 泛型模型定义统一响应结构。
创建统一成功响应函数。
定义业务异常基类和具体异常。
使用 @app.exception_handler() 注册全局异常处理器。
统一处理 HTTP 异常。
统一处理 Pydantic 参数验证异常。
统一处理未捕获的服务器异常。
使用 Axios 拦截器处理统一 API 响应。
使用中间件记录请求日志和请求 ID。
区分 HTTP 状态码与业务状态码。

对于一个长期维护的 FastAPI 项目，统一响应和异常处理能够显著降低前后端联调成本，也能让错误定位和日志分析更加规范。

【声明】本内容来自华为云开发者社区博主，不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息，否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@huaweicloud.com

点赞
收藏
关注作者

0/1000

抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认取消

加入云驻计划，成为创作者

华为云周边好礼
免费体验产品
特殊身份标识
线下官方门票
内部专家零距离
与10000+优质创作者共同成长

立即加入