FastAPI怎样通过中间件捕获所有未处理异常并格式化输出？

fastapi 中间件可统一处理未捕获异常并提升 api 稳定性。1. 创建自定义中间件函数 catch_exceptions_middleware，使用 try…except 捕获 httpexception 和其他异常；2. 注册该中间件至 fastapi 应用；3. 对 httpexception 返回其 detail 和 status_code，对其他异常记录日志并返回 500 响应；4. 可扩展错误响应格式，如添加 timestamp 和 path 字段；5. 异步任务中需单独处理异常，如在任务内捕获或使用 asyncio.gather；6. 使用 testclient 编写测试用例验证异常处理逻辑是否正确。通过这些步骤实现统一、详细的错误处理机制。

FastAPI 中间件可以有效地捕获未处理的异常，并以统一的格式输出，从而提升 API 的稳定性和可维护性。这主要通过自定义中间件来实现，在中间件中捕获异常，然后返回一个格式化的错误响应。

解决方案

1. 创建自定义中间件：定义一个中间件类或函数，用于捕获异常。
2. 注册中间件：将自定义中间件添加到 FastAPI 应用中。
3. 异常处理逻辑：在中间件中，使用 try…except 块捕获异常，并格式化输出。

如何处理不同类型的异常，并提供更详细的错误信息？

在中间件中，可以针对不同类型的异常进行处理。例如，可以区分 HTTPException 和其他类型的异常。对于 HTTPException，可以直接使用其 status_code 和 detail 属性。对于其他异常，可以记录异常信息，并返回一个通用的错误响应。

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")
# 模拟一个其他类型的异常
if item_id == 1:
raise ValueError("Invalid item ID")
return {"item_id": item_id}

这个例子中，catch_exceptions_middleware 中间件捕获了 HTTPException 和其他类型的异常。HTTPException 直接返回其错误信息，而其他异常则记录到日志中，并返回一个通用的 “Internal Server Error” 响应。通过日志记录，可以方便地调试和排查问题。

如何避免中间件影响性能？

异常处理本身会带来一定的性能开销。为了避免中间件影响性能，应该尽量减少 try…except 块中的代码量。只在必要的地方捕获异常，避免过度使用。另外，可以使用异步操作，避免阻塞主线程。

可以考虑使用更轻量级的异常处理方式，例如，只在生产环境中启用详细的错误日志记录，而在开发环境中直接抛出异常。

如何自定义错误响应的格式？

JSONResponse 可以自定义错误响应的格式。可以根据需要，添加额外的字段，例如 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)

在这个例子中，错误响应添加了 timestamp 和 path 字段，提供了更详细的错误信息。datetime.utcnow().isoformat() 用于生成 UTC 时间戳，request.url.path 用于记录请求的路径。

如何处理异步任务中的异常？

如果在 FastAPI 中使用了异步任务（例如，使用 asyncio.create_task 创建的任务），需要确保这些任务中的异常也被正确处理。一种方法是在任务中使用 try…except 块，并将异常信息传递给主线程。另一种方法是使用 asyncio.gather 来运行多个任务，并捕获 asyncio.gather 抛出的异常。

import asyncio
from fastapi import BackgroundTasks
async def my_task(item_id: int):
try:
# 模拟一个可能抛出异常的任务
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"}

在这个例子中，my_task 函数模拟了一个可能抛出异常的异步任务。如果任务失败，异常信息会被记录到日志中。create_task 函数使用 BackgroundTasks 来启动任务，确保任务在后台运行。

如何测试中间件的异常处理？

可以使用 FastAPI 的测试客户端来测试中间件的异常处理。可以编写测试用例，模拟各种异常情况，并验证中间件是否返回了正确的错误响应。

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键是否存在，不检查具体内容

这些测试用例验证了 read_item 函数在不同异常情况下的行为。test_read_item_invalid_id 验证了 HTTPException 的处理，test_read_item_value_error 验证了其他类型异常的处理。