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 的稳定性和可维护性。这主要通过自定义中间件来实现，在中间件中捕获异常，然后返回一个格式化的错误响应。
解决方案
- 创建自定义中间件：定义一个中间件类或函数，用于捕获异常。
- 注册中间件：将自定义中间件添加到 FastAPI 应用中。
- 异常处理逻辑：在中间件中，使用 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 验证了其他类型异常的处理。
