.jpg)
fastapi 中間件可統(tǒng)一處理未捕獲異常并提升 api 穩(wěn)定性。1. 創(chuàng)建自定義中間件函數(shù) catch_exceptions_middleware，使用 try…except 捕獲 httpexception 和其他異常；2. 注冊該中間件至 fastapi 應(yīng)用；3. 對 httpexception 返回其 detail 和 status_code，對其他異常記錄日志并返回 500 響應(yīng)；4. 可擴(kuò)展錯(cuò)誤響應(yīng)格式，如添加 timestamp 和 path 字段；5. 異步任務(wù)中需單獨(dú)處理異常，如在任務(wù)內(nèi)捕獲或使用 asyncio.gather；6. 使用 testclient 編寫測試用例驗(yàn)證異常處理邏輯是否正確。通過這些步驟實(shí)現(xiàn)統(tǒng)一、詳細(xì)的錯(cuò)誤處理機(jī)制。
FastAPI 中間件可以有效地捕獲未處理的異常，并以統(tǒng)一的格式輸出，從而提升 API 的穩(wěn)定性和可維護(hù)性。這主要通過自定義中間件來實(shí)現(xiàn)，在中間件中捕獲異常，然后返回一個(gè)格式化的錯(cuò)誤響應(yīng)。
解決方案
- 創(chuàng)建自定義中間件：定義一個(gè)中間件類或函數(shù)，用于捕獲異常。
- 注冊中間件：將自定義中間件添加到 FastAPI 應(yīng)用中。
- 異常處理邏輯：在中間件中，使用 try…except 塊捕獲異常，并格式化輸出。
如何處理不同類型的異常，并提供更詳細(xì)的錯(cuò)誤信息？
在中間件中，可以針對不同類型的異常進(jìn)行處理。例如，可以區(qū)分 HTTPException 和其他類型的異常。對于 HTTPException，可以直接使用其 status_code 和 detail 屬性。對于其他異常，可以記錄異常信息，并返回一個(gè)通用的錯(cuò)誤響應(yīng)。
from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse import logging app = FastAPI() # 配置日志 logging.basicConfig(level=logging.Error) logger = logging.getLogger(__name__) async def catch_exceptions_middleware(request: Request, call_next): try: return await call_next(request) except HTTPException as e: return JSONResponse({"error": e.detail}, status_code=e.status_code) except Exception as e: logger.exception(f"Unhandled exception: {e}") # 記錄完整的異常信息 return JSONResponse({"error": "internal Server Error"}, status_code=500) app.middleware("http")(catch_exceptions_middleware) @app.get("/items/{item_id}") async def read_item(item_id: int): if item_id == 0: raise HTTPException(status_code=400, detail="Item ID cannot be zero") # 模擬一個(gè)其他類型的異常 if item_id == 1: raise ValueError("Invalid item ID") return {"item_id": item_id}
這個(gè)例子中，catch_exceptions_middleware 中間件捕獲了 HTTPException 和其他類型的異常。HTTPException 直接返回其錯(cuò)誤信息，而其他異常則記錄到日志中，并返回一個(gè)通用的 “Internal Server Error” 響應(yīng)。通過日志記錄，可以方便地調(diào)試和排查問題。
如何避免中間件影響性能？
異常處理本身會帶來一定的性能開銷。為了避免中間件影響性能，應(yīng)該盡量減少 try…except 塊中的代碼量。只在必要的地方捕獲異常，避免過度使用。另外，可以使用異步操作，避免阻塞主線程。
可以考慮使用更輕量級的異常處理方式，例如，只在生產(chǎn)環(huán)境中啟用詳細(xì)的錯(cuò)誤日志記錄，而在開發(fā)環(huán)境中直接拋出異常。
如何自定義錯(cuò)誤響應(yīng)的格式？
JSONResponse 可以自定義錯(cuò)誤響應(yīng)的格式。可以根據(jù)需要，添加額外的字段，例如 timestamp、request_id 等。
from datetime import datetime async def catch_exceptions_middleware(request: Request, call_next): try: return await call_next(request) except HTTPException as e: return JSONResponse({ "timestamp": datetime.utcnow().isoformat(), "path": request.url.path, "error": e.detail }, status_code=e.status_code) except Exception as e: logger.exception(f"Unhandled exception: {e}") return JSONResponse({ "timestamp": datetime.utcnow().isoformat(), "path": request.url.path, "error": "Internal Server Error" }, status_code=500)
在這個(gè)例子中，錯(cuò)誤響應(yīng)添加了 timestamp 和 path 字段，提供了更詳細(xì)的錯(cuò)誤信息。datetime.utcnow().isoformat() 用于生成 UTC 時(shí)間戳，request.url.path 用于記錄請求的路徑。
如何處理異步任務(wù)中的異常？
如果在 FastAPI 中使用了異步任務(wù)（例如，使用 asyncio.create_task 創(chuàng)建的任務(wù)），需要確保這些任務(wù)中的異常也被正確處理。一種方法是在任務(wù)中使用 try…except 塊，并將異常信息傳遞給主線程。另一種方法是使用 asyncio.gather 來運(yùn)行多個(gè)任務(wù)，并捕獲 asyncio.gather 拋出的異常。
import asyncio from fastapi import BackgroundTasks async def my_task(item_id: int): try: # 模擬一個(gè)可能拋出異常的任務(wù) await asyncio.sleep(1) if item_id == 2: raise ValueError("Task failed") print(f"Task completed for item {item_id}") except Exception as e: logger.exception(f"Task failed: {e}") @app.post("/tasks") async def create_task(item_id: int, background_tasks: BackgroundTasks): background_tasks.add_task(my_task, item_id) return {"message": "Task started"}
在這個(gè)例子中，my_task 函數(shù)模擬了一個(gè)可能拋出異常的異步任務(wù)。如果任務(wù)失敗，異常信息會被記錄到日志中。create_task 函數(shù)使用 BackgroundTasks 來啟動任務(wù)，確保任務(wù)在后臺運(yùn)行。
如何測試中間件的異常處理？
可以使用 FastAPI 的測試客戶端來測試中間件的異常處理。可以編寫測試用例，模擬各種異常情況，并驗(yàn)證中間件是否返回了正確的錯(cuò)誤響應(yīng)。
from fastapi.testclient import TestClient client = TestClient(app) def test_read_item_invalid_id(): response = client.get("/items/0") assert response.status_code == 400 assert response.json() == {"error": "Item ID cannot be zero"} def test_read_item_value_error(): response = client.get("/items/1") assert response.status_code == 500 assert "error" in response.json() # 僅僅檢查error鍵是否存在，不檢查具體內(nèi)容
這些測試用例驗(yàn)證了 read_item 函數(shù)在不同異常情況下的行為。test_read_item_invalid_id 驗(yàn)證了 HTTPException 的處理，test_read_item_value_error 驗(yàn)證了其他類型異常的處理。
.png)
