FastAPI 异常处理实战：从 HTTPException 到全局处理器配置

FastAPI 异常处理的设计目标

FastAPI 基于 Starlette，默认会捕获未处理异常并返回堆栈或简单错误。在生产环境中，直接暴露 traceback 会导致安全风险，且无法为前端提供结构化的错误反馈。合理的异常处理体系应实现：统一错误响应格式、区分 HTTP 协议状态码与业务错误码、覆盖常规请求与 WebSocket 长连接，并在开发与生产环境安全切换日志输出。

第1步：使用 HTTPException 规范基础路由错误

对于参数校验失败、资源不存在等明确场景，直接在路由函数中抛出 HTTPException 是最直接的方式。通过 status_code 传递标准 HTTP 状态，detail 向用户说明原因，headers 可附加给前端的扩展字段（如推荐数据或链路追踪标识）。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    # 假设 item_db 为已初始化的数据源
    if item_id not in item_db:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "ItemID-Missing"}
        )
    return {"item": item_db[item_id]}

第2步：注册全局异常处理器统一响应格式

全局处理器是拦截所有未处理异常的入口。通过 @app.exception_handler 装饰器，可自定义返回结构。关键规则是先注册具体异常类型，后注册通用 Exception，否则具体类型会被通用处理器吞噬。生产环境应隐藏详细堆栈，仅开发环境保留以便排查。

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import os
import traceback
from typing import Optional

app = FastAPI()

# 统一错误响应模型
class ErrorResponse(BaseModel):
    code: int
    message: str
    detail: Optional[str] = None
    request_id: Optional[str] = None

# 1. 处理 HTTPException，保持原有状态码与 Headers
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content=ErrorResponse(
            code=exc.status_code,
            message=exc.detail,
            request_id=request.headers.get("X-Request-ID", "unknown")
        ).dict(),
        headers=exc.headers
    )

# 2. 兜底捕获所有未处理异常
@app.exception_handler(Exception)
async def universal_exception_handler(request: Request, exc: Exception):
    request_id = request.headers.get("X-Request-ID", "unknown")
    error_code = 500
    message = "Internal Server Error"
    
    if isinstance(exc, ValueError):
        error_code = 400
        message = "Invalid input value"
    
    # 开发环境打印完整堆栈，生产环境隐藏
    detail = traceback.format_exc() if os.getenv("ENV") == "development" else None
    
    return JSONResponse(
        status_code=error_code,
        content=ErrorResponse(
            code=error_code,
            message=message,
            detail=detail,
            request_id=request_id
        ).dict()
    )

第3步：定义并注册自定义业务异常

当业务逻辑出现特定状态（如余额不足、活动过期）时，使用标准 HTTP 状态码语义不准确。继承 Exception 创建自定义异常类，可携带业务错误码、提示信息及关联数据。随后注册专属处理器，将数据展开至响应体，前端可直接读取数据进行界面引导。

class BusinessError(Exception):
    def __init__(self, code: int, message: str, extra_data: dict = None):
        self.code = code
        self.message = message
        self.extra_data = extra_data or {}

class InsufficientBalanceError(BusinessError):
    def __init__(self, current_balance: float, required_amount: float):
        super().__init__(
            code=1001,
            message="Insufficient balance",
            extra_data={
                "current_balance": current_balance,
                "required_amount": required_amount
            }
        )

@app.exception_handler(BusinessError)
async def business_exception_handler(request: Request, exc: BusinessError):
    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "error": {
                "code": exc.code,
                "message": exc.message,
                **exc.extra_data
            }
        }
    )

@app.post("/purchase")
async def make_purchase(user_id: int, amount: float):
    user_balance = get_balance(user_id)
    if user_balance < amount:
        raise InsufficientBalanceError(
            current_balance=user_balance,
            required_amount=amount
        )
    return {"status": "success"}

第4步：处理 WebSocket 长连接异常

WebSocket 连接无法返回 JSON 响应，需通过关闭帧传递状态码与原因。捕获异常后应抛出 WebSocketException 并指定 RFC 标准码（如 1000 正常关闭、1008 政策违规），客户端据此决定重连或提示用户。

from fastapi import WebSocket, WebSocketException

VALID_TYPES = ["chat", "ping"]

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_json()
            if data.get("type") not in VALID_TYPES:
                raise WebSocketException(code=1008, reason="Invalid message type received")
    except WebSocketException as e:
        raise
    except Exception as e:
        raise WebSocketException(code=1011, reason=f"Internal error: {str(e)}")

开发注意事项与排错建议

• 避免过度捕获：不要使用 try...except Exception 包裹大范围业务代码，这会掩盖真实 Bug。仅捕获预期内且具备明确恢复或降级方案的异常。
• 完善日志记录：异常处理器中务必使用 logging.error(exc_info=True) 记录完整堆栈，并注入用户 ID、请求参数等上下文信息，便于事后追踪。
• 区分协议状态码与业务错误码：HTTPException.status_code 遵循 RFC 标准供网关过滤；自定义异常 code 用于前后端契约，两者配合使用。
• 编写测试用例：针对边界条件编写单元测试，模拟异常触发路径，验证响应结构与前端约定是否一致。

以上就是 FastAPI 异常处理配置的详细内容，更多关于 Python 后端框架最佳实践的资料请关注本站其它相关文章！