FastAPI 中介層（Middleware）— add_middleware() 使用方式

簡介

在 FastAPI 中，中介層（Middleware） 是介於請求（request）與回應（response）之間的可插拔程式碼。它可以在請求抵達路由之前、回應離開伺服器之前，執行統一的前置或後置處理。常見的應用包括 日誌紀錄、跨來源資源分享（CORS）設定、認證驗證、請求速率限制 等。

FastAPI 本身是基於 Starlette 建構，而 add_middleware() 正是 Starlette 提供的 API。透過它，我們可以在應用程式啟動時，輕鬆把自訂或第三方的 Middleware 加入到整個請求處理管線中。掌握 add_middleware() 的寫法與注意事項，對於打造可維護、可擴充的 Web API 極為關鍵。

---

核心概念

1. Middleware 的基本結構

在 Starlette/FastAPI 中，一個 Middleware 必須是一個 callable class，其 __init__ 接收 app（下一層的 ASGI 應用）以及可選的設定參數，__call__ 必須是一個 async 函式，接受 scope, receive, send 三個 ASGI 介面。

class SimpleLogMiddleware:
    def __init__(self, app):
        self.app = app                      # 下一層的 ASGI 應用

    async def __call__(self, scope, receive, send):
        # 只處理 HTTP 請求
        if scope["type"] == "http":
            method = scope["method"]
            path = scope["path"]
            print(f"[Log] {method} {path}")
        # 呼叫下一層
        await self.app(scope, receive, send)

2. add_middleware() 的語法

FastAPI（或 Starlette）的 add_middleware 方法接受兩個主要參數：

app.add_middleware(
    MiddlewareClass,            # 必須是可呼叫的 class
    **options                   # 任意關鍵字參數，會傳給 MiddlewareClass.__init__
)

Tip：options 只會傳給 Middleware 的 __init__，不會影響 __call__ 的簽名。

3. 為什麼使用 add_middleware() 而不是直接在 app = FastAPI() 時傳入 middleware=

• 可讀性：在大型專案中，Middleware 往往會散落在不同檔案。使用 add_middleware() 可以把所有註冊集中在 main.py，一目了然。
• 動態註冊：根據環境變數或設定檔，只在特定情況下加入某些 Middleware（例如測試環境不開啟速率限制）。
• 相容性：某些第三方套件（如 fastapi-limiter）只提供 MiddlewareClass，必須透過 add_middleware() 注入。

---

程式碼範例

範例 1️⃣：內建 CORS Middleware

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# 允許所有來源、所有方法、所有標頭
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

說明：CORSMiddleware 是 FastAPI 內建的 Middleware，只要在 add_middleware 時提供對應的設定即可。

範例 2️⃣：自訂請求日誌 Middleware（含額外參數）

import time
from fastapi import FastAPI

class RequestTimingMiddleware:
    def __init__(self, app, logger):
        self.app = app
        self.logger = logger               # 注入自訂 logger

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        start = time.time()
        await self.app(scope, receive, send)
        duration = (time.time() - start) * 1000  # ms
        method = scope["method"]
        path = scope["path"]
        self.logger.info(f"[Timing] {method} {path} - {duration:.2f}ms")

app = FastAPI()
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("myapp")

app.add_middleware(RequestTimingMiddleware, logger=logger)

說明：透過 add_middleware 把 logger 物件傳入 __init__，讓 Middleware 可以使用外部資源（如 logging、資料庫連線）。

範例 3️⃣：速率限制（Rate Limiting）Middleware（使用 aioredis）

import aioredis
from fastapi import FastAPI, HTTPException, status

class RateLimitMiddleware:
    def __init__(self, app, redis_url: str, limit: int, window: int):
        self.app = app
        self.redis_url = redis_url
        self.limit = limit          # 每個 window 允許的請求次數
        self.window = window        # 視窗秒數

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        client_ip = scope["client"][0]
        key = f"rl:{client_ip}"
        redis = await aioredis.from_url(self.redis_url)

        # 取得目前計數，若不存在則自動建立 TTL
        count = await redis.incr(key)
        if count == 1:
            await redis.expire(key, self.window)

        if count > self.limit:
            # 超過上限，直接回傳 429
            await send({
                "type": "http.response.start",
                "status": status.HTTP_429_TOO_MANY_REQUESTS,
                "headers": [(b"content-type", b"application/json")],
            })
            await send({
                "type": "http.response.body",
                "body": b'{"detail":"Rate limit exceeded"}',
            })
            await redis.close()
            return

        await self.app(scope, receive, send)
        await redis.close()

app = FastAPI()
app.add_middleware(
    RateLimitMiddleware,
    redis_url="redis://localhost:6379",
    limit=100,          # 每分鐘最多 100 次
    window=60,
)

說明：此範例示範 非同步 與外部資源（Redis）結合的情境，並在超過限制時直接回傳 429 Too Many Requests。

範例 4️⃣：全局錯誤捕獲 Middleware（轉換例外為 JSON 回應）

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

class ExceptionHandlerMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        try:
            await self.app(scope, receive, send)
        except Exception as exc:
            # 取得原始 request 以組合回應
            request = Request(scope)
            error_body = {
                "detail": str(exc),
                "path": request.url.path,
                "method": request.method,
            }
            response = JSONResponse(content=error_body, status_code=500)
            await response(scope, receive, send)

app = FastAPI()
app.add_middleware(ExceptionHandlerMiddleware)

說明：捕獲所有未處理的例外，統一回傳 JSON，對前端調試非常有幫助。

範例 5️⃣：條件式註冊（僅在開發環境啟用）

import os
from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

if os.getenv("ENV") == "production":
    # 只在正式環境強制 HTTPS
    app.add_middleware(HTTPSRedirectMiddleware)

說明：透過環境變數決定是否加入 Middleware，避免在開發時被不必要的重定向卡住。

---

常見陷阱與最佳實踐

陷阱	可能的結果	解決方案 / 最佳實踐
Middleware 順序錯誤	例如 CORS 放在自訂認證之後，導致瀏覽器仍收到 CORS 錯誤。	先加入最外層的 Middleware（如 CORS、HTTPSRedirect），再加入業務相關的。FastAPI 會依加入順序形成「管線」；越早加入的越先執行。
在 __call__ 中忘記 await self.app(...)	請求卡住、回應永遠不會送出，產生 504 超時。	確保在所有分支路徑最後都 await 下一層的 app，或在提前回應時直接使用 send。
使用同步 I/O（如 requests）於 async Middleware	事件迴圈被阻塞，導致整個服務效能下降。	使用非同步套件（httpx, aioredis, aiomysql），或在必要時把同步程式碼包在 run_in_threadpool。
在 Middleware 中直接拋出例外	例外不會被 FastAPI 的全局例外處理器捕獲，回傳 500 且無自訂訊息。	在 Middleware 內自行捕獲例外，或使用 ExceptionHandlerMiddleware 之類的統一處理層。
忘記關閉外部連線（Redis、DB）	連線資源泄漏，最終導致服務崩潰。	在 __call__ 完成後 務必關閉或釋放 連線；或使用 依賴注入 (Depends) 讓 FastAPI 自動管理生命週期。

最佳實踐總結

1. 保持 Middleware 輕量：僅負責單一職責（Single Responsibility）。
2. 使用型別註解：便利 IDE 自動補全與文件生成。
3. 加入日誌：在開發與除錯階段，日誌是定位問題的第一手資料。
4. 測試：使用 TestClient 撰寫單元測試，確保 Middleware 在不同情境下的行為如預期。
5. 文件化：在專案的 README 或 Wiki 中說明每個 Middleware 的目的與設定方式，降低新成員上手成本。

---

實際應用場景

場景	需要的 Middleware	為何使用 add_middleware()
跨域 API	CORSMiddleware	只要在 main.py 加入一次，即可全局支援前端跨域請求。
企業內部 API 需要 JWT 驗證	自訂 AuthMiddleware（讀取 Authorization Header、驗證 JWT）	透過 add_middleware(AuthMiddleware, secret_key="...") 把密鑰注入，避免在每個路由重複寫驗證程式。
流量高峰期的速率限制	RateLimitMiddleware（Redis + Token Bucket）	只在正式環境加入，開發環境可直接省略，提高測試效率。
統一錯誤回應格式	ExceptionHandlerMiddleware	在所有未捕獲例外時返回統一 JSON，前端只要寫一次錯誤處理程式。
HTTPS 強制	HTTPSRedirectMiddleware	在生產環境加入，確保所有 HTTP 請求被自動轉為 HTTPS，提升安全性。

---

總結

• add_middleware() 是 FastAPI（實際上是 Starlette）提供的 中介層註冊入口，可讓開發者在應用啟動階段，彈性加入任何符合 ASGI 規範的 Middleware。
• 順序、非同步安全、資源釋放 是實作時最常碰到的問題，掌握這些要點能避免效能瓶頸與意外錯誤。
• 透過 自訂 Middleware（如日誌、速率限制、統一錯誤處理）以及 內建 Middleware（CORS、HTTPSRedirect），我們可以把跨切面（cross‑cutting）關注點從業務路由中抽離，讓程式碼更乾淨、可測試、易維護。

實務建議：在新專案的 main.py 中先規劃好 全局 Middleware 的註冊清單，並使用環境變數或設定檔控制是否啟用。這樣不僅能保持程式碼結構清晰，也能在不同部署環境（開發、測試、正式）間快速切換，提升開發效率與部署彈性。

祝你在 FastAPI 的開發旅程中，利用好 Middleware，寫出更安全、更高效的 API！ 🚀