本文 AI 產出，尚未審核
FastAPI 中介層（Middleware） – 呼叫順序完整指南
簡介
在 FastAPI 這樣的非同步 Web 框架中，中介層 (Middleware) 扮演著「請求與回應」之間的過濾與加工工作。它們可以用來實作 認證、日誌、CORS、壓縮、錯誤處理 等功能，幾乎所有跨所有路由的需求都會透過 Middleware 來完成。
然而，當應用程式同時註冊多個 Middleware 時，呼叫順序 直接影響到功能是否正確、效能是否最佳，甚至會導致難以偵測的錯誤。本文將深入探討 FastAPI 中介層的執行流程、呼叫順序的決定因素，並提供實作範例、常見陷阱與最佳實踐，讓你在開發中能夠掌握每一層的「先後」與「作用範圍」。
核心概念
1. Middleware 的基本原理
FastAPI 的 Middleware 本質上是 ASGI（Asynchronous Server Gateway Interface）層級的函式。它接受 receive、send、scope 三個參數，並在 請求 (request) 進入 與 回應 (response) 出去 之間執行自訂邏輯。
from starlette.types import ASGIApp, Receive, Scope, Send
class SimpleMiddleware:
def __init__(self, app: ASGIApp):
self.app = app # 下游的 ASGI 應用
async def __call__(self, scope: Scope, receive: Receive, send: Send):
# 1️⃣ 請求進來前的處理
print("Before request")
await self.app(scope, receive, send) # 呼叫下游
# 2️⃣ 回應送出前的處理
print("After response")
重點：Middleware 以 堆疊 (stack) 方式組成，最先註冊的會 最外層 包住後續的 Middleware，呼叫順序就像「外層先進、內層先出」。
2. FastAPI 中 Middleware 的註冊方式
FastAPI 提供兩種常見方式：
| 方法 | 說明 | 何時使用 |
|---|---|---|
app.add_middleware() |
直接在 FastAPI 實例上加入，適合需要 依賴注入（如 BaseHTTPMiddleware）的情況 |
大多數情境 |
@app.middleware("http") 裝飾器 |
簡潔的函式式寫法，適合快速測試或簡單的前置/後置處理 | 小型或臨時需求 |
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
app = FastAPI()
# 方式一：使用 add_middleware
app.add_middleware(BaseHTTPMiddleware, dispatch=my_dispatch)
# 方式二：使用裝飾器
@app.middleware("http")
async def simple_middleware(request, call_next):
# 前置處理
response = await call_next(request) # 呼叫下游
# 後置處理
return response
3. 呼叫順序的決定因素
註冊順序
app.add_middleware()依照 程式碼執行的先後 加入堆疊，越早加入的越外層。@app.middleware裝飾的函式則會 在所有add_middleware之後 加入，且依照程式碼出現的先後順序排列。
BaseHTTPMiddleware與自訂ASGI類別的差異BaseHTTPMiddleware內部會 先將請求轉成 Starlette 的Request物件，再交給dispatch；因此它的「前置」與「後置」程式碼會在 同一層 執行。- 自訂
ASGI類別（如上例的SimpleMiddleware）則會直接 在await self.app(... )前後分別執行，更貼近原始的 ASGI 呼叫模型。
異步與同步的影響
- 異步 Middleware 必須使用
await呼叫下游；若在同步 Middleware 中呼叫異步下游，會觸發 事件迴圈錯誤。 - 為了避免此問題，FastAPI 推薦使用
BaseHTTPMiddleware（自動封裝為 async）或starlette.middleware系列。
- 異步 Middleware 必須使用
4. 呼叫順序圖解
┌─────────────────────────────────────┐
│ 1️⃣ 最外層 Middleware (最早 add) │
│ ┌───────────────────────────────┐ │
│ │ 2️⃣ 第二層 Middleware │ │
│ │ ┌─────────────────────────┐ │ │
│ │ │ 3️⃣ 第三層 Middleware │ │ │
│ │ │ (最內層, 最後 add) │ │ │
│ │ │ ┌───────────────────┐ │ │ │
│ │ │ │ FastAPI 路由 │ │ │ │
│ │ │ └───────────────────┘ │ │ │
│ │ └─────────────────────────┘ │ │
│ └───────────────────────────────┘ │
└─────────────────────────────────────┘
- 請求階段：從最外層往內層依序執行「前置」程式碼。
- 回應階段：從最內層往外層依序執行「後置」程式碼（即 先進後出 的堆疊模型）。
程式碼範例
以下提供 五個實用範例，說明不同情境下的 Middleware 設計與呼叫順序。
範例 1️⃣：最簡單的裝飾器式 Middleware（日誌）
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def log_middleware(request: Request, call_next):
# 前置：記錄請求資訊
print(f"[Log] {request.method} {request.url}")
response = await call_next(request) # 呼叫下游
# 後置：記錄回應狀態碼
print(f"[Log] status_code={response.status_code}")
return response
說明：此 Middleware 會最先被加入（若放在檔案最上方），因此它會是最外層，先捕捉所有請求與回應。
範例 2️⃣：使用 BaseHTTPMiddleware 實作簡易認證
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import HTTPException, Request
class AuthMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
token = request.headers.get("Authorization")
if token != "Bearer secret-token":
raise HTTPException(status_code=401, detail="Unauthorized")
# 前置處理結束，直接呼叫下游
response = await call_next(request)
return response
app.add_middleware(AuthMiddleware) # 先加入 → 最外層
說明：因為
AuthMiddleware先 被add_middleware，它會在所有後續 Middleware 之前檢查授權，確保未授權的請求不會進入其他層。
範例 3️⃣：自訂 ASGI Middleware（壓縮回應）
import gzip
from starlette.types import ASGIApp, Receive, Scope, Send
from fastapi import Response
class GzipMiddleware:
def __init__(self, app: ASGIApp, minimum_size: int = 500):
self.app = app
self.minimum_size = minimum_size
async def __call__(self, scope: Scope, receive: Receive, send: Send):
async def send_wrapper(message):
if message["type"] == "http.response.start":
headers = dict(message["headers"])
# 若回應大小足夠且客戶端支援 gzip，加入 Content-Encoding
if int(headers.get(b"content-length", b"0")) >= self.minimum_size:
message["headers"].append((b"content-encoding", b"gzip"))
if message["type"] == "http.response.body":
body = message.get("body", b"")
if body:
compressed = gzip.compress(body)
message["body"] = compressed
message["headers"] = [
(k, v) for k, v in message["headers"]
if k.lower() != b"content-length"
]
message["headers"].append(
(b"content-length", str(len(compressed)).encode())
)
await send(message)
await self.app(scope, receive, send_wrapper)
app.add_middleware(GzipMiddleware) # 加在 Auth 之後 → 內層
說明：此 Middleware 在回應送出前 進行 gzip 壓縮。因為它是 最後加入，所以在回應階段會 最先執行（最內層），確保壓縮發生在所有其他後置處理之後。
範例 4️⃣：錯誤捕捉 Middleware（全域例外處理）
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi.responses import JSONResponse
class ExceptionMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
try:
response = await call_next(request)
return response
except Exception as exc:
# 捕捉所有未處理例外，回傳 JSON 錯誤訊息
return JSONResponse(
status_code=500,
content={"detail": f"Server error: {str(exc)}"},
)
app.add_middleware(ExceptionMiddleware) # 最後加入 → 最外層
說明：把它加在 最後（最外層）可以保證不管哪一層拋出例外，都能被此 Middleware 捕捉。若放在較內層，外層的 Middleware 仍可能直接攔截例外，導致此層失效。
範例 5️⃣：跨域（CORS）與自訂 Header（示範呼叫順序）
from fastapi.middleware.cors import CORSMiddleware
# 1️⃣ CORS 必須在最外層，才能讓瀏覽器在 preflight 時即得到允許
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# 2️⃣ 自訂 Header Middleware（放在 CORS 之後）
class HeaderMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
response = await call_next(request)
response.headers["X-Powered-By"] = "FastAPI"
return response
app.add_middleware(HeaderMiddleware)
說明：若
HeaderMiddleware先於CORSMiddleware加入，瀏覽器的 preflight 可能會因缺少 CORS Header 而被阻擋。透過正確的加入順序，兩者皆能正常運作。
常見陷阱與最佳實踐
| 陷阱 | 說明 | 解決方式 |
|---|---|---|
| 順序寫錯，導致認證被繞過 | 把認證 Middleware 放在 最內層，其他 Middleware（如 CORS）先於它執行，可能讓未授權請求先返回 200。 | 先加入認證 Middleware，確保它是最外層。 |
在同步 Middleware 中直接 await |
會拋出 RuntimeError: Cannot use asyncio.run() from a non-async context。 |
使用 BaseHTTPMiddleware 或將整個 Middleware 定義為 async。 |
| 重複加入同類 Middleware | 會產生多次相同的 Header、壓縮或日誌，浪費資源。 | 檢查 app.user_middleware 或使用單例模式。 |
忘記在 call_next 前返回 |
若在前置處理中直接 return，下游路由不會被呼叫，導致 404 或無回應。 |
確保 只有在需要中斷流程時（如認證失敗）才直接回傳；否則一定要 await call_next(request)。 |
| 在回應階段改變已送出的 body | ASGI 的 send 只能一次送出完整的 body，後續修改不會生效。 |
在 send_wrapper 中 先攔截，再修改後再送出。 |
最佳實踐：
- 明確分層：把「安全」類 Middleware（認證、授權）放在最外層；「跨域」放在次外層；「日誌」與「計時」可放在最內層或外層，視需求決定。
- 使用
BaseHTTPMiddleware：除非需要直接操作 ASGI，否則建議使用它，因為它自動處理await、request轉型，降低錯誤率。 - 統一測試呼叫順序：使用
TestClient撰寫單元測試，檢查每個 Header、狀態碼、日誌是否符合預期。 - 避免過度堆疊：過多 Middleware 會增加每次請求的額外開銷，建議只保留必要的功能，或在同一 Middleware 中合併相關邏輯。
- 文件化 Middleware 順序：在專案的 README 或架構文件中列出 Middleware 的加入順序，方便新人快速了解整體流程。
實際應用場景
場景 1：企業內部 API 網關
- 需求：所有請求必須先經過 IP 白名單、JWT 認證，再由 日誌、計時、CORS 處理，最後返回壓縮回應。
- 實作：
# 1️⃣ IP 白名單（最外層）
app.add_middleware(IPWhitelistMiddleware)
# 2️⃣ JWT 認證
app.add_middleware(JWTAuthMiddleware)
# 3️⃣ 日誌與計時（同層或分別加入）
app.add_middleware(LoggingMiddleware)
app.add_middleware(TimingMiddleware)
# 4️⃣ CORS（在日誌之後，確保 preflight 會被允許）
app.add_middleware(CORSMiddleware, allow_origins=["https://example.com"])
# 5️⃣ Gzip 壓縮（最內層）
app.add_middleware(GzipMiddleware)
這樣的堆疊保證 安全檢查 必先完成，跨域 不會被日誌或計時攔截，壓縮 必在所有回應處理完畢後才執行。
場景 2：多租戶 SaaS 平台的請求隔離
- 需求：根據租戶 ID（從 Header 取得）切換 資料庫連線、日誌前綴，同時需要 全域錯誤捕捉。
- 實作：
app.add_middleware(TenantMiddleware) # 依租戶切換 DB，最外層
app.add_middleware(ExceptionMiddleware) # 捕捉所有錯誤，最外層
app.add_middleware(LoggingMiddleware) # 加入租戶前綴的日誌
TenantMiddleware必須最先執行，因為後續的 DB 操作、日誌都依賴正確的租戶上下文。
場景 3：前端開發環境的熱重載與偵錯
- 需求：在開發環境加入 自訂的 Debug Toolbar，同時保留 CORS、壓縮，但不影響正式環境。
- 實作：
if settings.DEBUG:
app.add_middleware(DebugToolbarMiddleware) # 只在開發時加入
app.add_middleware(CORSMiddleware, allow_origins=["*"])
app.add_middleware(GzipMiddleware)
透過條件式加入，確保 生產環境 不會帶入額外的除錯 Middleware，維持效能與安全。
總結
- Middleware 的呼叫順序 決定了請求與回應在整個應用程式中的「先後」與「可見」範圍。
- 最外層（最先
add_middleware）在 請求階段先執行前置程式碼，在 回應階段最後執行後置程式碼；相對的，最內層 在回應階段最先執行後置程式碼。 - 正確的 加入順序 能確保安全檢查、跨域、錯誤處理、壓縮等功能不會互相衝突。
- 使用
BaseHTTPMiddleware或 裝飾器式 Middleware 能降低非同步錯誤的風險，且更易於維護。 - 在實務開發中，先規劃好 Middleware 的層級，再依需求逐層加入，並透過測試驗證每一層的行為，是避免隱蔽 bug 的最佳策略。
掌握了 Middleware 的呼叫順序，你就能在 FastAPI 中構建 安全、可維護、效能優化 的 API 服務，讓專案在面對日益複雜的需求時，仍能保持彈性與可擴充性。祝開發順利！ 🚀