FastAPI 中介層（Middleware） – 自訂 Middleware 類別

簡介

在 Web 框架中，Middleware（中介層）是一段在請求（Request）與回應（Response）之間執行的程式碼。它可以攔截、修改、甚至拒絕請求，或在回應返回前加入統一的處理邏輯。FastAPI 內建支援 ASGI Middleware，使得我們可以輕鬆插入認證、日誌、跨來源資源共享（CORS）等功能。

對於 初學者 來說，了解如何自行撰寫 Middleware，不僅能加深對 ASGI 生命週期的認識，更能在實務專案中快速解決重複性的跨路由需求。本文將從概念說明、程式碼範例、常見陷阱與最佳實踐，一直到實際應用場景，完整帶你打造自訂 Middleware。

---

核心概念

1. Middleware 的運作時機

FastAPI 基於 Starlette，而 Starlette 再以 ASGI 為底層協定。當一個 HTTP 請求抵達時，ASGI server（如 Uvicorn）會依序呼叫註冊的 Middleware，形成一條管道（pipeline）：

[Client] → [Middleware A] → [Middleware B] → … → [FastAPI router] → [Response] → … → [Middleware B] → [Middleware A] → [Client]

每個 Middleware 必須實作一個 __call__ 方法，接受 scope, receive, send 三個參數，且必須 await 下一層的呼叫，才能讓請求繼續往下傳遞。

2. 建立自訂 Middleware 類別的基本步驟

1. 繼承 BaseHTTPMiddleware（或直接實作 ASGI Callable）。
2. 在 dispatch 方法中處理請求，並呼叫 call_next(request) 取得下游回應。
3. 回傳或修改回應，最後交給 FastAPI。

Tip：若需要存取 request.state、response.headers 等，請使用 BaseHTTPMiddleware，它已幫你封裝好 Request 物件。

3. 何時使用自訂 Middleware？

• 統一日誌：記錄每筆請求的路徑、方法、耗時等。
• 自訂認證：在路由前檢查 API 金鑰或 JWT。
• 回應壓縮：根據 Accept-Encoding 自動 gzip。
• 錯誤統一格式：捕捉未處理的例外，回傳 JSON 錯誤訊息。

---

程式碼範例

以下示範 5 個常見的自訂 Middleware，皆以 Python 為語言，使用 ````python` 標記。

1️⃣ 基礎範例：計時 Middleware

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

class TimerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)          # 呼叫下游
        process_time = (time.time() - start_time) * 1000
        # 在回應 Header 加入處理時間
        response.headers["X-Process-Time-ms"] = str(round(process_time, 2))
        return response

app = FastAPI()
app.add_middleware(TimerMiddleware)

說明：call_next 會返回 Response，我們在 Header 中加入 X-Process-Time-ms，前端即可直接看到每筆請求的耗時。

2️⃣ 日誌 Middleware（支援 async logger）

import logging
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

logger = logging.getLogger("uvicorn.access")

class AccessLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        logger.info(f"▶ {request.method} {request.url.path}")
        response = await call_next(request)
        logger.info(f"◀ {request.method} {request.url.path} - {response.status_code}")
        return response

app = FastAPI()
app.add_middleware(AccessLogMiddleware)

說明：利用 uvicorn.access logger，將請求與回應分別記錄，方便排查問題。

3️⃣ API 金鑰驗證 Middleware

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

API_KEY = "my-secret-key"

class ApiKeyMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        token = request.headers.get("X-API-Key")
        if token != API_KEY:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid API Key",
            )
        return await call_next(request)

app = FastAPI()
app.add_middleware(ApiKeyMiddleware)

說明：在每筆請求的 Header 必須帶入 X-API-Key，否則直接拋出 401 錯誤，避免路由內重複寫驗證程式。

4️⃣ GZIP 壓縮 Middleware（僅在支援的情況下壓縮回應）

import gzip
from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

class GzipMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response: Response = await call_next(request)
        accept_encoding = request.headers.get("Accept-Encoding", "")
        if "gzip" in accept_encoding.lower():
            body = gzip.compress(response.body)
            response.body = body
            response.headers["Content-Encoding"] = "gzip"
            response.headers["Content-Length"] = str(len(body))
        return response

app = FastAPI()
app.add_middleware(GzipMiddleware)

說明：若客戶端支援 gzip，就把回應內容壓縮，減少網路傳輸量。注意：此範例僅示範概念，實務上建議使用 starlette.middleware.gzip.GZipMiddleware。

5️⃣ 統一錯誤回傳 Middleware（捕捉未處理例外）

import json
from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.status import HTTP_500_INTERNAL_SERVER_ERROR

class ExceptionHandlerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            return await call_next(request)
        except Exception as exc:
            # 可以在此記錄 log
            payload = {"detail": "Internal Server Error", "error": str(exc)}
            return Response(
                content=json.dumps(payload),
                status_code=HTTP_500_INTERNAL_SERVER_ERROR,
                media_type="application/json",
            )

app = FastAPI()
app.add_middleware(ExceptionHandlerMiddleware)

說明：將未捕捉的例外轉成統一的 JSON 回應，前端只要依賴一個格式即可處理所有錯誤。

---

常見陷阱與最佳實踐

陷阱	為什麼會發生	解決方式
忘記 await call_next(request)	Middleware 成為同步函式，導致請求卡住不返回	必須使用 await，或改寫成 同步 Middleware（較少見）
在 Middleware 中直接拋出 HTTPException	若未捕捉，會被 Starlette 的例外處理器捕獲，可能產生不一致的錯誤格式	建議在 Middleware 內自行回傳 Response，或在全局例外處理器統一格式化
重複加入同一 Middleware	會造成效能浪費，甚至 Header 被覆寫兩次	檢查 app.user_middleware，確保只加入一次
在 dispatch 中執行大量 I/O（例如 DB 連線）	會阻塞事件迴圈，降低整體併發量	使用 async I/O，或將重度運算搬到背景任務（Celery、RQ）
忘記在回傳前設定 Content-Length（壓縮或修改 body）	部分瀏覽器或代理會卡住連線	手動更新 response.headers["Content-Length"]，或使用 starlette.responses.StreamingResponse

最佳實踐

1. 保持 Middleware 輕量：只處理與請求/回應直接相關的事務，其他業務邏輯最好放在路由或依賴注入（Depends）中。
2. 使用 request.state 共享資料：若需要在多個 Middleware 或路由間傳遞資料，可將資訊寫入 request.state，避免全局變數。
3. 避免在回應 Header 中寫入敏感資訊：Header 會被瀏覽器或中介代理看到，僅存放非機密資料。
4. 測試 Middleware：利用 TestClient 撰寫單元測試，確保在不同 Header、方法、狀態碼下仍能正確執行。
5. 設定 Middleware 執行順序：先加入的 Middleware 會先執行（請求階段）且最後回傳（回應階段），依需求排列順序。

---

實際應用場景

1. 企業內部 API Gateway

   • 需求：所有內部服務必須透過 API 金鑰驗證，同時紀錄每筆請求的耗時與使用者 ID。
   • 解決：自訂兩個 Middleware：ApiKeyMiddleware + TimerMiddleware，再把使用者 ID 放入 request.state.user_id，供下游路由使用。

2. 多租戶 SaaS 平台

   • 需求：根據請求的 Host 標頭，切換不同資料庫連線。
   • 解決：在 Middleware 中解析 Host，設定 request.state.db 為相應的資料庫 Session，路由只需要 Depends(get_db) 即可取得正確連線。

3. 前端需要即時顯示 API 呼叫耗時

   • 需求：回傳的 JSON 必須包含 process_time_ms 欄位。
   • 解決：在 TimerMiddleware 中把耗時寫入 response.body（先 decode、再 encode），或直接在 Header 中傳遞，前端自行取用。

4. 安全合規（PCI DSS）

   • 需求：所有回傳的敏感欄位（如信用卡號）必須被遮蔽。
   • 解決：自訂 MaskSensitiveDataMiddleware，在 dispatch 完成後檢查 response.media_type == "application/json"，使用正則表達式或字典遍歷遮蔽特定鍵。

---

總結

自訂 FastAPI Middleware 是提升應用程式一致性、可維護性與安全性的關鍵技巧。透過本文的概念說明與五個實作範例，你應該已掌握：

• Middleware 的生命週期與 BaseHTTPMiddleware 的使用方式。
• 如何在 請求前 做驗證、日誌、計時，或在 回應前 進行壓縮、錯誤統一格式化。
• 常見的 陷阱（忘記 await、同步阻塞）與 最佳實踐（保持輕量、使用 request.state、測試）。
• 在 企業級、多租戶、安全合規 等實務情境中，如何運用自訂 Middleware 解決問題。

只要遵循「輕量、可組合、統一錯誤處理」的原則，你的 FastAPI 專案將更具彈性，也更容易在未來加入新的全域功能。祝你在開發過程中玩得開心，寫出乾淨、可維護的 API！