你的API在用户面前“裸奔”了吗？-平芜编程栈

第一部分：异常处理不是备选项，而是必选项

把API想象成一家餐厅。用户点餐（发送请求），厨房处理（服务端逻辑），最后上菜（返回响应）。异常处理是什么？就是当厨房发现“鱼卖完了”或者“客人对海鲜过敏”时，服务员如何得体地告知顾客，并给出替代方案，而不是直接把锅摔了，或者扔给顾客一张看不懂的后厨采购单（Python traceback）。

我刚用FastAPI那会儿，也偷懒过，觉得有默认错误页面就行。结果呢？前端同事天天找我要错误码对照表，测试同学报的Bug描述模糊不清，线上出了问题定位慢如蜗牛。血的教训告诉我们：异常处理必须和业务逻辑同步设计，甚至要更早考虑。

🛡️ 第二部分：HTTPException，用好它但别依赖它

FastAPI提供了HTTPException，这是最直接、最常用的异常抛出方式。它就像一个标准化的“错误通知单”。

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): 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]}

看这段代码，status_code告诉前端这是什么类型的错误（404找不到了），detail给人类看的原因，headers里还能塞点给机器看的额外信息。是不是很像服务员说：“抱歉先生，您点的这道菜（item_id）今天售罄了（404），这是我们推荐的相似菜品（headers里可以放推荐）”。

但是！千万别以为只用HTTPException就万事大吉了。想象一下，你餐厅的后厨着火了（服务器内部错误），或者客人拿了一张假钞来付款（请求数据根本不符合格式），这时候只靠服务员说“菜没了”显然不够。我们需要更强大的机制。

🔧 第三部分：打造你的“异常消防队”——全局异常处理器

全局异常处理器（Exception Handler）就是你API大楼里的自动消防系统和万能服务员。任何没被特定处理的异常，最终都会落到这里，由它统一格式，友好返回。

from fastapi import FastAPI, Request from fastapi.responses import JSONResponse import traceback app = FastAPI() # 1. 先定义一个标准的错误响应模型 class ErrorResponse(BaseModel): code: int message: str detail: Optional[str] = None request_id: Optional[str] = None # 用于链路追踪 # 2. 捕获所有未处理异常的“总闸” @app.exception_handler(Exception) async def universal_exception_handler(request: Request, exc: Exception): # 获取请求ID，便于追踪（假设从中间件或header传入） request_id = request.headers.get("X-Request-ID", "unknown") # 这里可以根据exc的类型进行更精细的分类 error_code = 500 # 默认内部错误 message = "Internal Server Error" if isinstance(exc, ValueError): error_code = 400 message = "Invalid input value" # ... 可以添加更多类型判断 # 在生产环境，detail可能不返回具体堆栈，开发环境可以返回 import os 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. 专门处理HTTPException，覆盖FastAPI默认行为 @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 )

这个“消防队”厉害在哪？首先，它抓住了所有Exception，确保没有异常会“裸奔”出去。其次，它把错误响应格式标准化了，前端永远知道会收到{"code": ..., "message": ...}这样的结构。最后，它还区分了开发和生成环境，开发时给你详细堆栈debug，生产环境则隐藏细节保证安全。

这里有个我踩过的大坑：异常处理器的注册顺序很重要！如果你先注册了通用的Exception处理器，再注册HTTPException处理器，那么HTTPException也会被通用的抓住，你就无法对它进行特殊定制了。所以，通常要先注册具体的，再注册通用的。

🎨 第四部分：自定义异常——让业务错误清晰明了

业务逻辑里的错误，比如“用户余额不足”、“活动已结束”，用404或400虽然也行，但语义不精确。这时候，就需要自定义异常。

# 定义自己的业务异常类 class BusinessError(Exception): def __init__(self, code: int, message: str, extra_data: dict = None): self.code = code # 业务错误码，如 1001 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 } ) class ActivityExpiredError(BusinessError): def __init__(self, activity_id: str, expire_time: str): super().__init__( code=1002, message="Activity has expired", extra_data={"activity_id": activity_id, "expire_time": expire_time} ) # 为自定义业务异常注册处理器 @app.exception_handler(BusinessError) async def business_exception_handler(request: Request, exc: BusinessError): return JSONResponse( status_code=422, # 或用200，但body里表明错误，看前端约定 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: # 抛出业务异常，而不是简单的HTTP 400 raise InsufficientBalanceError( current_balance=user_balance, required_amount=amount ) # ... 购买逻辑

这样做的好处巨大！前端看到错误码1001，就知道是余额不足，并且直接从extra_data里拿到当前余额和所需金额，可以立刻在界面上友好提示：“您的余额为XX元，还需充值YY元”。这体验，比干巴巴的“请求失败”好了一万倍。

⚡ 第五部分：WebSocketException——实时通道的优雅关闭

WebSocket是长连接，异常处理方式和HTTP不太一样。你不能返回一个JSON响应，而是需要优雅地关闭连接并发送原因。

from fastapi import WebSocket, WebSocketException @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: # 抛出WebSocketException，指定关闭码和原因 raise WebSocketException( code=1008, # 1008表示政策违规 reason="Invalid message type received" ) # ... 处理消息 except WebSocketException as e: # 这里其实raise之后，FastAPI会帮你关闭连接 raise except Exception as e: # 其他未知异常，也以WebSocketException形式关闭 raise WebSocketException(code=1011, reason=f"Internal error: {str(e)}")

WebSocket关闭码是有标准的，比如1000表示正常关闭，1008表示政策违规。用好这些代码，能让客户端明确知道连接为什么断开，从而做出相应处理（比如重连、提示用户等）。