FastAPI 全局异常处理最佳实践：自定义异常、统一响应、兜底处理

在 FastAPI 项目里如果每个接口都写一堆try...except项目刚开始可能还能接受接口一多就会变得很乱每个接口返回的错误格式不一致。controller、service、crud 的职责边界变模糊。前端不好统一处理错误提示。日志记录分散线上问题不好排查。系统异常可能直接暴露给前端存在安全风险。更推荐的方式是controller / service / crud 只写正常业务流程业务异常统一raise最后由 FastAPI 的全局异常处理器统一转换成响应。本文会用一套比较贴近实际项目的写法整理 FastAPI 全局异常处理的设计思路和代码模板。一、本文适合什么场景本文适合下面这些场景FastAPI 项目想统一接口返回格式。不想在每个接口里重复写try...except。希望业务异常、参数异常、系统异常分开处理。希望 controller、service、crud 分层更清楚。希望项目里有一套可以复用的异常处理模板。示例不绑定具体 ORM。你用 SQLAlchemy、Tortoise ORM、SQLModel 或其他数据库工具都可以按这个思路改造。二、为什么需要全局异常处理先看一种不太推荐的写法router.post(/users) async def create_user_api(form: UserCreate): try: user await create_user(form) return { success: True, code: 0, message: 创建成功, data: user, } except Exception as exc: return JSONResponse( status_code500, content{ success: False, code: 50000, message: str(exc), }, )这段代码的问题是controller 既处理请求又处理异常响应职责太重。except Exception太宽泛容易把业务异常和系统异常混在一起。str(exc)可能把数据库错误、文件路径、内部实现细节暴露给前端。每个接口都这样写会产生大量重复代码。更好的做法是router.post(/users) async def create_user_api(form: UserCreate): user await create_user(form) return success_response(datauser, message创建用户成功)controller 只关心正常流程异常交给全局异常处理器。一句话总结正常流程写在业务代码里异常响应统一交给 exception handler。三、FastAPI 官方异常机制FastAPI 本身已经提供了异常处理机制主要有几个常用点机制作用HTTPExceptionFastAPI 内置异常适合返回 HTTP 错误RequestValidationError请求参数校验失败时触发app.exception_handler(...)注册全局异常处理器app.add_exception_handler(...)另一种集中注册异常处理器的方式FastAPI 官方文档里也强调HTTPException是一个普通的 Python 异常只是带了 HTTP 响应相关信息。你可以在接口函数、service、依赖函数、工具函数里raise它请求会被中断然后由异常处理器生成响应。项目里如果需要统一响应格式就可以通过app.exception_handler(...)或app.add_exception_handler(...)覆盖默认异常响应。四、异常分类项目中不要把所有错误都当成一种异常。一般可以分成四类。1. 业务异常业务异常指的是代码运行没有问题但是业务规则不允许继续执行。比如用户不存在用户名已存在密码错误余额不足库存不足订单状态不允许取消这种异常适合定义成自己的业务异常类例如BusinessException。2. 参数异常参数异常指的是请求参数不符合接口要求。比如必填字段没传字段类型错误字符串长度不符合要求枚举值不合法路径参数类型错误FastAPI 结合 Pydantic 做参数校验时如果请求参数不合法会触发RequestValidationError。3. 认证和权限异常这类异常一般和登录、token、角色权限有关。比如未登录token 过期token 无效没有接口访问权限没有资源操作权限可以直接使用HTTPException(status_code401)或HTTPException(status_code403)也可以根据项目习惯封装成业务异常。4. 系统异常系统异常是代码运行过程中出现的非预期错误。比如数据库连接失败Redis 连接失败第三方接口超时文件读写失败未捕获的 Python 异常系统异常不应该把原始异常信息直接返回给前端。更合理的方式是日志里记录真实异常前端只收到统一的错误提示。五、推荐项目结构可以按下面这种结构组织代码app/ ├── main.py ├── core/ │ ├── response.py │ ├── exceptions.py │ └── exception_handlers.py ├── api/ │ └── user.py ├── services/ │ └── user_service.py ├── crud/ │ └── user_crud.py └── schemas/ └── user.py核心文件说明文件作用response.py统一成功和失败响应格式exceptions.py定义自定义业务异常exception_handlers.py定义全局异常处理器main.py注册异常处理器services/写业务流程业务不成立时raisecrud/写数据库操作尽量保持简单六、统一响应格式为了让前端更好处理建议接口统一返回类似结构{ success: false, code: 40001, message: 用户名已存在, data: null }字段说明字段含义success业务是否成功code业务状态码message错误提示或成功提示data业务数据这里要注意区分两个概念status_codeHTTP 状态码比如 200、400、401、403、404、422、500。code业务状态码比如 40001 表示用户名已存在。不要把 HTTP 状态码和业务状态码混在一起。HTTP 状态码表示请求层面的结果业务 code 表示项目内部约定的业务结果。七、代码模板下面给出一套可以直接放进项目里的基础模板。1. 统一响应工具# app/core/response.py from typing import Any def success_response( data: Any None, message: str success, code: int 0, ) - dict: return { success: True, code: code, message: message, data: data, } def error_response( message: str, code: int, data: Any None, ) - dict: return { success: False, code: code, message: message, data: data, }2. 自定义业务异常# app/core/exceptions.py from http import HTTPStatus class BusinessException(Exception): def __init__( self, message: str, code: int 40000, status_code: int HTTPStatus.BAD_REQUEST, ): self.message message self.code code self.status_code status_code这个异常类只保存异常信息不负责返回JSONResponse。字段说明字段作用message返回给前端的错误提示code业务错误码status_codeHTTP 状态码3. 全局异常处理器# app/core/exception_handlers.py import logging from fastapi import FastAPI, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from starlette.exceptions import HTTPException as StarletteHTTPException from app.core.exceptions import BusinessException from app.core.response import error_response logger logging.getLogger(__name__) async def business_exception_handler( request: Request, exc: BusinessException, ) - JSONResponse: return JSONResponse( status_codeexc.status_code, contenterror_response( codeexc.code, messageexc.message, ), ) async def validation_exception_handler( request: Request, exc: RequestValidationError, ) - JSONResponse: return JSONResponse( status_code422, contenterror_response( code42200, message请求参数校验失败, dataexc.errors(), ), ) async def http_exception_handler( request: Request, exc: StarletteHTTPException, ) - JSONResponse: return JSONResponse( status_codeexc.status_code, contenterror_response( codeexc.status_code, messagestr(exc.detail), ), ) async def unknown_exception_handler( request: Request, exc: Exception, ) - JSONResponse: logger.exception(未处理的系统异常: %s %s, request.method, request.url) return JSONResponse( status_code500, contenterror_response( code50000, message服务器内部错误, ), ) def register_exception_handlers(app: FastAPI) - None: app.add_exception_handler(BusinessException, business_exception_handler) app.add_exception_handler(RequestValidationError, validation_exception_handler) app.add_exception_handler(StarletteHTTPException, http_exception_handler) app.add_exception_handler(Exception, unknown_exception_handler)这里用了StarletteHTTPException是因为 FastAPI 底层基于 Starlette。这样不仅能处理 FastAPI 主动抛出的HTTPException也能处理 Starlette 内部或相关扩展抛出的 HTTP 异常。4. 在 main.py 中注册# app/main.py from fastapi import FastAPI from app.core.exception_handlers import register_exception_handlers app FastAPI() register_exception_handlers(app)这样全局异常处理器就注册好了。八、分层怎么写全局异常处理不是单独存在的它要配合项目分层一起使用。1. controller 层controller 负责接收请求、调用 service、返回结果。controller 不要写大量业务判断也不要到处写try...except。# app/api/user.py from fastapi import APIRouter from app.core.response import success_response from app.schemas.user import UserCreate from app.services.user_service import create_user router APIRouter(prefix/users, tags[用户]) router.post() async def create_user_api(form: UserCreate): user await create_user(form) return success_response(datauser, message创建用户成功)controller 这层保持简单异常不在这里展开处理。2. service 层service 负责业务规则判断。业务不成立时直接抛出业务异常。# app/services/user_service.py from app.core.exceptions import BusinessException from app.crud.user_crud import get_user_by_username, save_user from app.schemas.user import UserCreate async def create_user(form: UserCreate): exists_user await get_user_by_username(form.username) if exists_user: raise BusinessException( message用户名已存在, code40001, status_code400, ) return await save_user(form)这段代码的业务流程很清楚查询用户名是否存在 - 已存在就抛业务异常 - 不存在就创建用户不需要在 service 里直接返回JSONResponse。3. crud 层crud 负责数据库操作尽量保持简单。# app/crud/user_crud.py async def get_user_by_username(username: str): return await User.filter(usernameusername).first() async def save_user(form): return await User.create(**form.model_dump())如果你使用的是 Pydantic v1可以把model_dump()换成dict()return await User.create(**form.dict())crud 层一般不写复杂业务判断。除非是数据库唯一约束、事务回滚、连接异常这类和数据访问强相关的逻辑否则业务规则建议放在 service 层。九、兜底异常处理为什么重要很多项目一开始只处理业务异常但没有处理未知异常。这样线上一旦出现未预期错误可能会返回很不稳定的响应甚至暴露内部信息。兜底异常处理器的作用是后端日志记录真实异常。前端收到统一的错误格式。避免把数据库错误、文件路径、堆栈信息暴露给用户。推荐写法async def unknown_exception_handler( request: Request, exc: Exception, ) - JSONResponse: logger.exception(未处理的系统异常: %s %s, request.method, request.url) return JSONResponse( status_code500, contenterror_response( code50000, message服务器内部错误, ), )不推荐这样写return JSONResponse( status_code500, content{message: str(exc)}, )原因很简单str(exc)不是给前端看的它可能包含数据库表名、SQL、文件路径、第三方服务返回的敏感信息。十、什么时候才需要 try...except全局异常处理不代表项目里完全不能写try...except。适合写try...except的场景包括调用第三方接口需要把超时转换成业务异常。操作文件、消息队列、Redis 等外部资源。需要做失败补偿比如回滚、释放资源。需要记录更详细的上下文日志。例如调用支付服务async def call_payment_api(order_id: int): try: return await payment_client.pay(order_id) except TimeoutError as exc: raise BusinessException( message支付服务超时请稍后重试, code50010, status_code503, ) from exc这里有两个关键点捕获异常后不要直接吞掉。使用raise ... from exc保留异常链方便排查原始错误。十一、不同异常应该返回什么状态码可以参考下面这个表场景HTTP 状态码说明参数校验失败422 或 400FastAPI 默认常用 422未登录401没有有效身份认证无权限403已登录但无权限资源不存在404查询对象不存在业务规则不通过400比如用户名已存在、余额不足第三方服务不可用503比如支付服务超时未知系统异常500服务器内部错误业务状态码可以自己约定比如业务 code含义0成功40000通用业务错误40001用户名已存在40100未登录40300无权限42200参数校验失败50000服务器内部错误具体编号不重要重要的是团队内部保持一致。十二、常见错误1. 每个接口都写 try...except这样会导致重复代码很多而且每个接口的错误格式容易不一致。更好的方式是业务错误抛BusinessException统一异常处理器负责返回响应。2. service 里直接返回 JSONResponseservice 层不应该关心 HTTP 响应。它应该只关心业务流程。不推荐async def create_user(form): if exists_user: return JSONResponse(...)推荐async def create_user(form): if exists_user: raise BusinessException(message用户名已存在, code40001)3. 把所有异常都返回 500用户名已存在、参数错误、权限不足都不是 500。500 表示服务器内部错误不适合表示普通业务失败。4. 把原始异常直接返回给前端错误信息应该分两层日志里记录真实异常。前端只返回安全、稳定、可理解的提示。5. crud 层写大量业务判断crud 层负责数据访问service 层负责业务规则。把大量业务判断放在 crud 层会让代码后期很难复用。十三、完整流程回顾整个请求流程可以理解成前端请求 - controller 接收参数 - service 执行业务流程 - crud 查询或写入数据库 - 如果业务不成立service raise BusinessException - exception handler 捕获异常 - 返回统一 JSON 响应职责可以总结成下面这张表层级职责是否建议处理异常controller接收请求、调用 service不建议大量try...exceptservice写业务流程和业务判断可以raise BusinessExceptioncrud数据库操作尽量保持简单exception handler统一转换异常响应专门处理异常logger记录真实错误记录系统异常细节十四、总结FastAPI 全局异常处理的重点不是“把所有错误都捕获掉”而是让项目的错误处理有清晰边界。推荐原则controller 只写接口入口。service 写业务流程业务不成立就raise。crud 只写数据访问不混入 HTTP 响应。自定义业务异常只保存错误信息。exception handler 统一把异常转换成响应。未知异常要兜底并记录日志。最终目标是业务代码保持 basic flow异常逻辑集中处理前端响应格式稳定线上问题可以通过日志排查。参考资料FastAPI 官方文档Handling ErrorsFastAPI 官方文档HTTPException Reference