FastAPI 中如何使用 Middleware 實現(xiàn)請求日志和耗時統(tǒng)計?
在日常開發(fā)中,我們經(jīng)常需要統(tǒng)計每個請求的耗時、記錄訪問日志,便于監(jiān)控接口性能和排查問題。在 FastAPI 中,這種需求的最佳實踐方式就是 —— 使用中間件(Middleware)!
本文將帶你一步步實現(xiàn)一個通用的請求耗時統(tǒng)計中間件,并自動記錄日志(方法、路徑、耗時等信息)。
什么是 Middleware?
Middleware(中間件)是介于請求進入與響應返回之間的一段可插拔邏輯,可以用來處理:
- 日志記錄
- 請求耗時
- 權限校驗
- 跨域處理(CORS)
- 請求頻率限制(Rate Limit)等
FastAPI 提供了便捷的方式來自定義中間件,核心是通過裝飾器或注冊機制封裝函數(shù)。
實現(xiàn)一個耗時統(tǒng)計中間件
我們下面寫一個中間件,來統(tǒng)計每個請求從進入到響應返回所花費的時間,并將日志打印出來:
middlewares/request_timer.py
import time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
import logging
# 設置日志輸出
logger = logging.getLogger("request_logger")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
formatter = logging.Formatter('[%(asctime)s] %(levelname)s: %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
class RequestTimerMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
start_time = time.time()
response = await call_next(request) # 執(zhí)行請求處理鏈
process_time = time.time() - start_time
formatted_time = f"{process_time * 1000:.2f}ms"
logger.info(f"{request.method} {request.url.path} - 耗時: {formatted_time}")
# 可以添加到響應頭中返回
response.headers["X-Process-Time"] = formatted_time
return response
如何在項目中啟用中間件?
只需在創(chuàng)建 FastAPI 應用實例時添加:
main.py
from fastapi import FastAPI
from middlewares.request_timer import RequestTimerMiddleware
app = FastAPI()
# 添加中間件
app.add_middleware(RequestTimerMiddleware)
# 示例路由
@app.get("/ping")
async def ping():
return {"message": "pong"}
訪問接口 /ping 后你將看到如下日志輸出:
[2025-04-29 10:30:00] INFO: GET /ping - 耗時: 2.33ms
可擴展的點
你還可以在中間件中擴展更多功能,比如:
- 記錄響應狀態(tài)碼
- 日志落盤或輸出到日志平臺(如 ELK)
- 日志中包含 IP、User-Agent、用戶信息等
- 將日志結構化為 JSON 格式(便于日志平臺解析)
總結
通過中間件的方式,我們實現(xiàn)了:
- 請求耗時統(tǒng)計
- 請求路徑和方法記錄
- 響應頭中加入處理時間
這是構建高質量 Web API 服務的必備實踐,尤其對 運維監(jiān)控和性能分析 非常有幫助。