FastAPI 中介層（Middleware）── Request 攔截與修改

---

簡介

在 Web 框架 中，中介層（Middleware） 扮演著「過濾器」的角色：每一次 HTTP 請求在到達路由處理函式前，都會先通過一或多個中介層。利用這個機制，我們可以在請求進入應用程式之前完成 驗證、日誌、統計、跨域設定 等工作，甚至直接 修改或替換 請求內容。

對於 FastAPI 這類以 Starlette 為底層的非同步框架而言，中介層的設計既簡潔又彈性，讓開發者只需要撰寫一個符合「ASGI」協定的可呼叫物件（callable），即可在全域或路由層級攔截請求。掌握這項技巧，不僅能提升程式碼的可重用性，還能在 安全性、效能、可觀測性 等方面為專案加分。

以下本文將從概念說明、實作範例、常見陷阱與最佳實踐，逐步帶領你建立 Request 攔截與修改 的中介層，讓你在 FastAPI 專案中即刻上手。

---

核心概念

1. Middleware 的基本結構

FastAPI 的中介層本質上是 ASGI 中間件，其呼叫簽名為：

async def __call__(self, scope: dict, receive: Callable, send: Callable) -> None:
    ...

• scope：包含請求的 meta 資訊（method、path、headers 等）。
• receive：用來接收客戶端傳來的訊息（如請求 Body）。
• send：用來回傳回應給客戶端。

在 FastAPI 中，我們通常使用 函式型 或 類別型 兩種寫法，框架會自動幫你包裝成符合 ASGI 的物件。

2. 為什麼要攔截與修改 Request

• 統一驗證：例如在每個請求頭部加上 JWT token，或檢查 API key。
• 自訂 Header：在所有回應自動加入 X-Request-ID、Server 等資訊。
• 請求 Body 轉換：把前端傳來的 snake_case 參數轉成 camelCase，或在解析 JSON 前先做清理。
• 日誌與監控：記錄每筆請求的耗時、IP、User‑Agent 等。

3. 中介層的執行順序

FastAPI 會依照 註冊順序（app.add_middleware）執行中介層，先加入的先「外層」執行，最後加入的則是「內層」：

Client -> Middleware A -> Middleware B -> Route Handler -> Middleware B -> Middleware A -> Client

因此，若你需要 先驗證後記錄，就必須把驗證中介層放在記錄中介層之前。

---

程式碼範例

以下提供 5 個實用範例，從最簡單的請求攔截到較進階的 Body 重新包裝，皆附上說明註解。

範例 1️⃣：簡單的請求日誌 Middleware

from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """在每筆請求前後印出基本資訊與耗時"""
    start_time = time.time()
    # 取得 HTTP 方法與路徑
    method = request.method
    path = request.url.path
    print(f"[Log] {method} {path} - 開始處理")
    
    # 呼叫下一層（實際的路由處理）
    response = await call_next(request)
    
    # 計算耗時
    duration = (time.time() - start_time) * 1000
    print(f"[Log] {method} {path} - 完成 ({duration:.2f}ms)")
    return response

重點：call_next 會傳回 Response 物件，若不呼叫就會 阻斷 請求流程。

範例 2️⃣：自動加入 X-Request-ID Header

import uuid
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    """為每筆請求產生唯一 ID，並寫入回應 Header"""
    request_id = str(uuid.uuid4())
    # 先把 request_id 放到 request.state，讓路由可取用
    request.state.request_id = request_id
    
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

技巧：request.state 是一個 任意屬性容器，可在同一個請求的不同階段共享資訊。

範例 3️⃣：驗證 API Key（Header）

from fastapi import FastAPI, Request, HTTPException, status

app = FastAPI()
VALID_API_KEY = "my-secret-key"

@app.middleware("http")
async def verify_api_key(request: Request, call_next):
    """檢查 X-API-Key 是否正確，錯誤則直接回傳 401"""
    api_key = request.headers.get("X-API-Key")
    if api_key != VALID_API_KEY:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API Key"
        )
    # 驗證通過，繼續下游
    return await call_next(request)

注意：在中介層直接拋出 HTTPException，FastAPI 會自動轉成對應的 HTTP 回應。

範例 4️⃣：修改 JSON Body（將 snake_case 轉為 camelCase）

import json
import re
from fastapi import FastAPI, Request
from starlette.responses import JSONResponse

app = FastAPI()

def snake_to_camel(s: str) -> str:
    """snake_case → camelCase"""
    parts = s.split('_')
    return parts[0] + ''.join(p.title() for p in parts[1:])

@app.middleware("http")
async def convert_body(request: Request, call_next):
    """只在 Content-Type 為 application/json 時，將 body 欄位名稱改為 camelCase"""
    if request.headers.get("content-type", "").startswith("application/json"):
        # 讀取原始 body（只能讀一次，需自行保存）
        body_bytes = await request.body()
        if body_bytes:
            body = json.loads(body_bytes)
            new_body = {snake_to_camel(k): v for k, v in body.items()}
            # 建立一個新的 Request，帶入改過的 body
            async def receive():
                return {"type": "http.request", "body": json.dumps(new_body).encode()}
            request = Request(request.scope, receive)
    # 繼續下游
    response = await call_next(request)
    return response

關鍵：await request.body() 只能讀一次，所以若要修改後繼續傳遞，必須重新建立 receive 來提供新的 body。

範例 5️⃣：類別型 Middleware – 計算全域請求速率（Rate Limiting）

from fastapi import FastAPI, Request, HTTPException, status
from starlette.middleware.base import BaseHTTPMiddleware
import time

class RateLimiterMiddleware(BaseHTTPMiddleware):
    """簡易的每分鐘 60 次限制（IP 為單位）"""
    def __init__(self, app, max_per_minute: int = 60):
        super().__init__(app)
        self.max_per_minute = max_per_minute
        self.access_log = {}   # {ip: [timestamp, ...]}

    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        now = time.time()
        timestamps = self.access_log.get(client_ip, [])
        # 移除已過期的時間點
        timestamps = [ts for ts in timestamps if now - ts < 60]
        if len(timestamps) >= self.max_per_minute:
            raise HTTPException(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                detail="Rate limit exceeded"
            )
        timestamps.append(now)
        self.access_log[client_ip] = timestamps
        # 繼續處理請求
        response = await call_next(request)
        return response

app = FastAPI()
app.add_middleware(RateLimiterMiddleware, max_per_minute=30)

說明：使用 BaseHTTPMiddleware 可以更方便地寫類別型中介層，且 dispatch 只需要關心 request 與 call_next。

---

常見陷阱與最佳實踐

陷阱	為何會發生	解決方案 / 最佳做法
讀取 Body 後無法再使用	await request.body() 只能讀一次，之後 call_next 會收到空 body。	在需要修改 Body 時，重新建立 receive 函式或使用 Starlette 的 Request 重新包裝。
中介層內拋出未捕獲的例外	若未捕獲，FastAPI 仍會轉成 500，但日誌可能不完整。	使用 try/except 包住 call_next，自行記錄錯誤或回傳自訂錯誤訊息。
中介層順序錯誤	先記錄後驗證會產生大量無效日誌。	按需求先 驗證 → 授權 → 日誌 → 回應加工 的順序註冊。
共享可變狀態	若在全域變數中保存請求相關資訊，會在多協程環境下產生競爭條件。	使用 request.state（每個請求獨立）或 ContextVar。
阻塞 I/O	在中介層內使用同步的資料庫/網路呼叫會阻塞事件迴圈。	使用 async 版的驅動或在中介層外部使用 run_in_threadpool。

進階最佳實踐

1. 最小化工作負載：中介層應只負責「跨切面」的事務，具體業務邏輯仍交給路由或服務層。
2. 可測試性：將中介層的核心邏輯抽成純函式，單元測試時可以直接呼叫而不必啟動整個 FastAPI 應用。
3. 設定化：將可變參數（如 API Key、速率上限）放入 環境變數 或 設定檔，避免硬編碼。
4. 記錄結構化日誌：使用 loguru、structlog 等套件，把請求 ID、耗時、IP 等欄位寫成 JSON，方便後續分析。

---

實際應用場景

場景	可能採用的 Middleware	目的
企業內部 API	驗證 JWT、加入公司標頭、IP 白名單	確保安全、統一追蹤
公開 SaaS 服務	Rate Limiting、API Key 驗證、CORS 設定	防止濫用、符合跨域需求
微服務間通訊	追蹤 trace-id、自動重試失敗請求	觀測分散式交易、提升可靠性
前端統一錯誤處理	捕捉未處理例外、回傳標準化錯誤格式	前端只需要處理一套錯誤結構
資料清理與轉換	Body 轉換（snake → camel、過濾敏感欄位）	降低路由層的資料前處理負擔

舉例：假設你在開發一個 多租戶平台，每筆請求必須帶有 X-Tenant-ID，且根據租戶不同套用不同的資料庫連線。只需要在中介層 擷取 X-Tenant-ID，存入 request.state，之後的路由或依賴注入即可根據這個值切換資料庫，整個流程不會在每個路由重複寫相同程式碼。

---

總結

• Middleware 是 FastAPI 中處理跨切面需求的核心機制，Request 攔截與修改 能讓你在進入路由前完成驗證、日誌、Header 加工、Body 轉換等工作。
• 實作上，既可以使用 函式型 (@app.middleware("http"))；也可以使用 類別型 (BaseHTTPMiddleware) 以獲得更高的彈性。
• 常見的陷阱包括 Body 只能讀一次、順序錯誤、同步阻塞，只要遵守 「最小化工作負載、保持非同步、使用 request.state」 的原則，就能寫出既安全又易維護的中介層。
• 在實務上，從 API 金鑰驗證、請求追蹤、速率限制 到 資料格式統一，都能透過中介層一次解決，讓路由程式碼保持乾淨、專注於業務本身。

掌握了 Request 攔截與修改 的技巧，你的 FastAPI 專案將更具 可觀測性、可擴充性與安全性。祝開發順利，快把這些範例搬到自己的專案裡試試看吧！