FastAPI 中介層（Middleware）

主題：Logging Middleware

---

簡介

在 Web 應用程式的開發與維運過程中，日誌（log） 是不可或缺的資訊來源。它不只協助開發者在除錯時快速定位問題，也能在系統上線後提供監控、分析與安全稽核的依據。FastAPI 作為一套高效能的 ASGI 框架，提供了彈性的 中介層（Middleware） 機制，讓我們能在請求（request）與回應（response）之間插入自訂程式碼，實作統一的日誌紀錄。

本篇文章將帶你一步一步了解 Logging Middleware 的概念與實作，從最簡單的請求時間記錄，到結合結構化日誌、異步寫檔與敏感資訊過濾等進階技巧，幫助你在 FastAPI 專案中快速建立可靠且可擴充的日誌系統。

---

核心概念

1. Middleware 的運作原理

FastAPI 基於 ASGI（Asynchronous Server Gateway Interface）規範，所有的請求都會經過一條 callable chain。Middleware 本質上是一個接受 request、呼叫 call_next(request)，再處理 response 的函式或類別。

async def simple_middleware(request: Request, call_next):
    # 1. 在此處理進入的 request
    response = await call_next(request)   # 呼叫下游的路由或其他 middleware
    # 2. 在此處理離開的 response
    return response

重點：Middleware 必須是 異步（async） 的，才能與 FastAPI 的非阻塞特性相容。

---

2. 為什麼要使用 Logging Middleware

需求	若不使用 Middleware 的結果	使用 Middleware 的好處
統一格式	每個路由自行寫 log，格式容易不一致	中介層一次設定，所有路由自動遵守
請求追蹤	必須在每個端點手動加入追蹤碼	中介層自動捕捉 request_id、IP、方法等
效能分析	只能在個別端點測量，難以彙總	中介層可一次記錄 執行時間、狀態碼
安全審計	敏感資訊可能遺漏	中介層可集中過濾與加密

---

3. 建立基本的 Logging Middleware

以下示範一個最簡單的日誌中介層，會記錄請求方法、路徑、狀態碼以及處理時間。

# logging_middleware.py
import time
import logging
from fastapi import Request, FastAPI

logger = logging.getLogger("uvicorn.access")   # 使用 uvicorn 預設 logger

async def logging_middleware(request: Request, call_next):
    start_time = time.time()
    # 取得請求資訊
    method = request.method
    url = str(request.url)

    # 呼叫下游
    response = await call_next(request)

    # 計算耗時
    process_time = (time.time() - start_time) * 1000  # ms
    status_code = response.status_code

    # 輸出日誌
    logger.info(
        f"{method} {url} - {status_code} - {process_time:.2f}ms"
    )
    return response

# FastAPI 應用程式
app = FastAPI()
app.middleware("http")(logging_middleware)

@app.get("/hello")
async def hello():
    return {"msg": "Hello World"}

說明

• app.middleware("http") 為 FastAPI 提供的裝飾器，可直接將函式註冊為 HTTP 中介層。
• 我們使用 uvicorn.access logger，這樣日誌會與伺服器的存取日誌合併，方便統一管理。

---

4. 結構化日誌（JSON）

在大型系統或使用 ELK、Grafana Loki 等集中式日誌平台時，JSON 結構化日誌 更易於搜尋與分析。下面示範如何把日誌改寫成 JSON 格式。

# structured_logging_middleware.py
import time
import json
import logging
from fastapi import Request, FastAPI

structured_logger = logging.getLogger("structured")
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))  # 直接輸出純文字
structured_logger.addHandler(handler)
structured_logger.setLevel(logging.INFO)

async def structured_logging_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = (time.time() - start) * 1000

    log_data = {
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "method": request.method,
        "path": request.url.path,
        "query": dict(request.query_params),
        "status": response.status_code,
        "duration_ms": round(duration, 2),
        "client_ip": request.client.host,
    }
    structured_logger.info(json.dumps(log_data))
    return response

app = FastAPI()
app.middleware("http")(structured_logging_middleware)

重點：

• 只要把字典 log_data 以 json.dumps 轉成字串，即可讓日誌系統直接解析。
• request.client.host 取得客戶端 IP，若在反向代理（如 Nginx）後，需自行從 X-Forwarded-For 抽取。

---

5. 敏感資訊過濾與 Request Body 記錄

在某些情境（例如 API 金流、登入）需要記錄 請求 Body，但同時必須避免將密碼、金鑰等資訊寫入日誌。以下示範一個可過濾敏感欄位的 Middleware。

# body_logging_middleware.py
import time
import json
import logging
from fastapi import Request, FastAPI
from starlette.datastructures import MutableHeaders

logger = logging.getLogger("body_logger")
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler())

SENSITIVE_FIELDS = {"password", "access_token", "refresh_token"}

def mask_sensitive(data: dict) -> dict:
    """將敏感欄位的值改為 ***"""
    return {
        k: ("***" if k in SENSITIVE_FIELDS else v)
        for k, v in data.items()
    }

async def body_logging_middleware(request: Request, call_next):
    # 讀取 request body（只能讀一次，需要重新放回）
    body_bytes = await request.body()
    try:
        body = json.loads(body_bytes) if body_bytes else {}
    except json.JSONDecodeError:
        body = {"raw": body_bytes.decode(errors="ignore")}

    # 過濾敏感資訊
    safe_body = mask_sensitive(body)

    # 記錄請求資訊
    logger.info(
        json.dumps({
            "method": request.method,
            "path": request.url.path,
            "body": safe_body,
            "client_ip": request.client.host,
        })
    )

    # 必須重新建立一個 Request 讓下游仍能讀取 body
    async def receive():
        return {"type": "http.request", "body": body_bytes}

    request = Request(request.scope, receive)

    response = await call_next(request)
    return response

app = FastAPI()
app.middleware("http")(body_logging_middleware)

@app.post("/login")
async def login(username: str, password: str):
    # 實作略
    return {"msg": "login success"}

說明

• await request.body() 只能讀一次，讀完後必須把原始 bytes 再包回 Request，否則下游路由會得到空的 body。
• mask_sensitive 函式負責將密碼等欄位隱蔽，確保日誌不會洩漏。

---

常見陷阱與最佳實踐

陷阱	描述	解決方案
同步阻塞	在 Middleware 中使用 time.sleep() 或同步檔案寫入會阻塞事件迴圈，導致所有請求排隊	改用 await asyncio.sleep() 或使用 非阻塞的 logger（如 uvicorn、loguru）
Body 只能讀一次	直接呼叫 await request.body() 後，下游路由無法再取得請求內容	如上例所示，讀完後重新建立 receive 方法或使用 request.stream() 逐塊讀取
過度日誌	記錄過多資訊（如完整的 query string、header）會使日誌體積暴增，影響效能與儲存成本	只保留關鍵欄位，敏感資訊務必過濾；可根據環境變數（dev / prod）調整日誌等級
未考慮反向代理	直接使用 request.client.host 會得到代理伺服器 IP，而非真實使用者 IP	從 X-Forwarded-For、X-Real-IP 取得原始 IP，並在部署文件中說明信任的代理清單
例外未捕捉	Middleware 若在 call_next 前後拋出例外，會導致伺服器回傳 500，且日誌不完整	使用 try/except 包住 call_next，在 except 區塊內記錄例外資訊並重新拋出或回傳自訂回應

最佳實踐

1. 統一日誌格式：採用 JSON 或 key‑value 形式，讓後續的 log aggregation 工具（ELK、Loki）能直接解析。
2. 分層日誌等級：開發環境使用 DEBUG，上線環境僅保留 INFO / ERROR。
3. 非阻塞寫檔：若需要寫檔，使用 aiofiles 或將日誌交給外部服務（如 Fluent Bit）處理。
4. 結合 Request ID：使用 uuid4 或 X-Request-ID 產生唯一識別碼，於所有日誌中帶入，方便跨服務追蹤。
5. 測試覆蓋：為 Middleware 撰寫單元測試，確保在高併發、異常情況下仍能正確記錄。

---

實際應用場景

場景	為何需要 Logging Middleware	可能的實作方式
API 監控儀表板	需要即時顯示每個端點的請求量、成功率、平均延遲	使用結構化日誌 + Prometheus exporter（把日誌轉成 metrics）
安全稽核	追蹤敏感操作（如帳號刪除、金流請求），保留完整的操作紀錄	結合 Request ID、使用者 ID、操作時間與 IP，寫入審計資料庫
錯誤追蹤	當服務拋出例外時，快速定位是哪個請求導致	在 Middleware 捕捉例外，記錄 traceback 與 request 資訊，再送到 Sentry
多服務鏈路追蹤	微服務架構下，請求會經過多個服務，需要跨服務的請求鏈路	在 Middleware 中注入 traceparent（W3C Trace Context），讓每個服務都傳遞同一個 trace ID
性能基準測試	想要比較不同版本 API 的效能差異	記錄每筆請求的毫秒耗時，匯出 CSV 供統計分析

---

總結

Logging Middleware 是 FastAPI 中最常見且最具價值的中介層之一。透過它，我們能在 單一位置 完成：

1. 統一且結構化的日誌輸出
2. 請求與回應的全程追蹤（包括執行時間、狀態碼、IP 等）
3. 敏感資訊的安全過濾
4. 跨服務的追蹤與監控

實作時只需要掌握 異步、body 只能讀一次 以及 日誌等級/格式 的設計原則，即可避免常見陷阱，打造出既 高效 又 可靠 的日誌系統。希望本篇文章的概念說明與範例程式碼，能讓你在自己的 FastAPI 專案中快速上手，並根據實際需求持續優化與擴充。祝開發順利，系統穩定！