你的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_code404, detailItem 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_codeerror_code, contentErrorResponse( codeerror_code, messagemessage, detaildetail, request_idrequest_id ).dict() ) # 3. 专门处理HTTPException覆盖FastAPI默认行为 app.exception_handler(HTTPException) async def http_exception_handler(request: Request, exc: HTTPException): return JSONResponse( status_codeexc.status_code, contentErrorResponse( codeexc.status_code, messageexc.detail, request_idrequest.headers.get(X-Request-ID, unknown) ).dict(), headersexc.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__( code1001, messageInsufficient 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__( code1002, messageActivity 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_code422, # 或用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_balanceuser_balance, required_amountamount ) # ... 购买逻辑这样做的好处巨大前端看到错误码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( code1008, # 1008表示政策违规 reasonInvalid message type received ) # ... 处理消息 except WebSocketException as e: # 这里其实raise之后FastAPI会帮你关闭连接 raise except Exception as e: # 其他未知异常也以WebSocketException形式关闭 raise WebSocketException(code1011, reasonfInternal error: {str(e)})WebSocket关闭码是有标准的比如1000表示正常关闭1008表示政策违规。用好这些代码能让客户端明确知道连接为什么断开从而做出相应处理比如重连、提示用户等。 第六部分避坑指南与进阶思考 1. 不要过度捕获异常别动不动就用try...except Exception把一大段业务逻辑包起来。这会隐藏真正的Bug。只捕获你预期中可能发生的、并且你知道如何处理的异常。 2. 日志日志日志异常处理器里一定要记日志而且要记录完整的堆栈信息和请求上下文用户ID、请求参数等。用logging.error(exc_infoTrue)。这是你事后排查问题的唯一指望。 3. 区分返回状态码status_code和业务错误码error_codeHTTP状态码是给HTTP协议和网关看的如404, 500。业务错误码是你和前端约定的具体错误含义如1001余额不足。两者可以结合使用。 4. 考虑使用Starlette的异常处理基类FastAPI基于Starlettefrom starlette.exceptions import HTTPException和FastAPI的略有不同。如果你需要更底层的控制可以研究一下。 5. 测试你的异常处理