FastAPI:(6)錯誤處理
由于CSDN無法展示「漸構」的「#d,#e,#t,#c,#v,#a」標簽,推薦訪問我個人網站進行閱讀:Hkini
「漸構展示」如下:
#c 概述 文章概念關系
graph TDA[錯誤處理] --> B[HTTPException]A --> C[自定義異常類]A --> D[異常處理器]D --> E[注冊處理器]E --> F[@app.exception_handler]B --> G[狀態碼]B --> H[detail]B --> I[headers]C --> J[UnicornException]D --> K[RequestValidationError]K --> L[請求體]K --> M[覆蓋默認處理器]D --> N[復用默認處理器]N --> O[fastapi.exception_handlers]
#d 錯誤處理
錯誤處理是指在應用運行過程中對可能發生的異常或錯誤進行捕獲、統一響應和返回格式化錯誤信息的機制。FastAPI 支持對內建異常(如 HTTPException
)的處理,以及自定義異常類型的注冊和響應方式。良好的錯誤處理機制可以提升 API 的魯棒性、用戶體驗和調試效率。
重要特征:
- 特征1:基于
HTTPException
或自定義異常類
FastAPI 默認支持fastapi.HTTPException
,也允許定義自己的異常類和異常處理器。 - 特征2:響應內容結構化,可自定義狀態碼、消息和返回體
錯誤響應可包括status_code
、detail
、headers
等字段。 - 特征3:通過
@app.exception_handler
注冊處理器
可針對特定異常類型定義處理邏輯,統一捕獲并生成響應。 - 特征4:與路徑操作函數邏輯解耦
使主業務代碼更簡潔清晰,錯誤處理集中統一。
#e 訂單查詢異常處理(正例)
現象:
在一個電商平臺中,用戶根據訂單 ID 查詢訂單詳情。若訂單不存在,則拋出 HTTPException(404)
,返回“訂單未找到”的錯誤響應。
特征對比
特征 | 是否滿足 |
---|---|
使用 HTTPException 或自定義異常 | ? 使用內建 HTTPException |
響應結構規范(狀態碼、消息) | ? 設置 404 + detail |
捕獲邏輯清晰明確 | ? 異常直接由業務邏輯觸發 |
與主業務邏輯解耦 | ? 查詢失敗即拋異常,無需后續處理判斷 |
from fastapi import FastAPI, HTTPExceptionapp = FastAPI()fake_orders = {"123": "訂單123內容", "456": "訂單456內容"}@app.get("/order/{order_id}")
async def get_order(order_id: str):if order_id not in fake_orders:raise HTTPException(status_code=404, detail="訂單未找到")return {"order_id": order_id, "content": fake_orders[order_id]}
HTTPException
是額外包含了和 API 有關數據的常規 Python 異常。因為是 Python 異常,所以不能 return
,只能 raise
。
如在調用「路徑操作函數」里的工具函數時,觸發了
HTTPException
,FastAPI 就不再繼續執行_路徑操作函數_中的后續代碼,而是立即終止請求,并把HTTPException
的 HTTP 錯誤發送至客戶端。
觸發
HTTPException
時,可以用參數detail
傳遞任何能轉換為 JSON 的值,不僅限于str
。還支持傳遞dict
、list
等數據結構。FastAPI 能自動處理這些數據,并將之轉換為 JSON。
#e return
返回錯誤信息(反例)
現象:
當用戶輸入非法請求時,系統返回如下內容:
return {"error": "invalid input"}
并未使用 HTTP 狀態碼或異常處理機制。
特征對比
特征 | 是否滿足 |
---|---|
使用 HTTPException 或自定義異常 | ? 直接 return,未拋異常 |
響應結構標準 | ? 未設置狀態碼,結構不規范 |
捕獲邏輯統一 | ? 所有錯誤需開發者手動處理 |
與主業務解耦 | ? 錯誤處理混雜在主邏輯中,易出錯 |
#e 自定義響應頭
有些場景下要為 HTTP 錯誤添加自定義響應頭。例如,出于某些方面的安全需要。一般情況下可能不會需要在代碼中直接使用響應頭。但對于某些高級應用場景,還是需要添加自定義響應頭:
from fastapi import FastAPI, HTTPExceptionapp = FastAPI()items = {"foo": "The Foo Wrestlers"}@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):if item_id not in items:raise HTTPException(status_code=404,detail="Item not found",headers={"X-Error": "There goes my error"}, # 添加響應頭)return {"item": items[item_id]}
#e 自定義異常處理器
添加自定義處理器,要使用 Starlette 的異常工具。假設要觸發的自定義異常叫作 UnicornException
。且需要 FastAPI 實現全局處理該異常。此時,可以用 @app.exception_handler()
添加自定義異常控制器:
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponseclass UnicornException(Exception):def __init__(self, name: str):self.name = nameapp = FastAPI()@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):return JSONResponse(status_code=418,content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},)@app.get("/unicorns/{name}")
async def read_unicorn(name: str):if name == "yolo":raise UnicornException(name=name)return {"unicorn_name": name}
請求 /unicorns/yolo
時,路徑操作會觸發 UnicornException
。
但該異常將會被 unicorn_exception_handler
處理。
接收到的錯誤信息清晰明了,HTTP 狀態碼為 418
,JSON 內容如下:
{"message": "Oops! yolo did something. There goes a rainbow..."}
#e 覆蓋請求驗證異常
「請求中包含無效數據」時,FastAPI 內部會觸發 RequestValidationError
。該異常也內置了默認異常處理器。
覆蓋默認異常處理器時需要導入 RequestValidationError
,并用 @app.excption_handler(RequestValidationError)
裝飾異常處理器。這樣,異常處理器就可以接收 Request
與異常。
from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPExceptionapp = FastAPI()@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):return PlainTextResponse(str(exc.detail), status_code=exc.status_code)@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):return PlainTextResponse(str(exc), status_code=400)@app.get("/items/{item_id}")
async def read_item(item_id: int):if item_id == 3:raise HTTPException(status_code=418, detail="Nope! I don't like 3.")return {"item_id": item_id}
#e 覆蓋 HTTPException
處理器
只為錯誤返回純文本響應,而不是返回 JSON 格式的內容。
from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPExceptionapp = FastAPI()@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):return PlainTextResponse(str(exc.detail), status_code=exc.status_code)@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):return PlainTextResponse(str(exc), status_code=400)@app.get("/items/{item_id}")
async def read_item(item_id: int):if item_id == 3:raise HTTPException(status_code=418, detail="Nope! I don't like 3.")return {"item_id": item_id}
#e RequestValidationError
的請求體
RequestValidationError
包含其接收到的無效數據請求的 body
。
開發時,可以用這個請求體生成日志、調試錯誤,并返回給用戶。
from fastapi import FastAPI, Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModelapp = FastAPI()@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):return JSONResponse(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),)class Item(BaseModel):title: strsize: int@app.post("/items/")
async def create_item(item: Item):return item
發送無效item
{"title": "towel","size": "XL"
}
收到422,Error: Unprocessable Entity的body
響應體。
```json
{"detail": [{"type": "int_parsing","loc": ["body","size"],"msg": "Input should be a valid integer, unable to parse string as an integer","input": "XL"}],"body": {"title": "towel","size": "XL"}
}
#e 復用FastAPI異常處理器
FastAPI 支持先對異常進行某些處理,然后再使用 FastAPI 中處理該異常的默認異常處理器。從 fastapi.exception_handlers
中導入要復用的默認異常處理器,可以在處理異常之后再復用默認的異常處理器。
from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import (http_exception_handler,request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPExceptionapp = FastAPI()@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):print(f"OMG! An HTTP error!: {repr(exc)}")return await http_exception_handler(request, exc)@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):print(f"OMG! The client sent invalid data!: {exc}")return await request_validation_exception_handler(request, exc)@app.get("/items/{item_id}")
async def read_item(item_id: int):if item_id == 3:raise HTTPException(status_code=418, detail="Nope! I don't like 3.")return {"item_id": item_id}