本文 AI 產出，尚未審核

FastAPI 中介層（Middleware） – Response 攔截與修改

簡介

在 FastAPI 中，中介層（Middleware） 是一段在請求（Request）抵達路由處理函式之前、或回應（Response）離開應用程式之前執行的程式碼。雖然大多數教學會聚焦於 Request 的驗證或日誌記錄，Response 的攔截與修改同樣重要，尤其在以下情境：

  • 統一回傳格式（例如加入 code、message、data 欄位）
  • 加密、壓縮或加入自訂的 HTTP Header
  • 記錄回應時間、內容長度或錯誤碼供監控系統使用

本篇文章將從概念說明、實作範例、常見陷阱與最佳實踐，逐步帶你掌握 FastAPI 中介層的 Response 攔截與修改 技巧，適合剛踏入 FastAPI 的新手，也能幫助已有基礎的開發者提升專案的可維護性與一致性。

核心概念

1. Middleware 的執行流程

FastAPI（底層使用 Starlette）在收到 HTTP 請求時，會依序執行 所有已註冊的 Middleware，每個 Middleware 必須是一個 callable，接受 request: Request 以及 call_next（呼叫下一個 Middleware 或路由處理函式）的兩個參數，並回傳一個 Response 物件。 

Client → Middleware A → Middleware B → Route Handler → Middleware B (返回) → Middleware A (返回) → Client
  • 前置階段：在 call_next 之前的程式碼可用來檢查或修改 Request。
  • 後置階段：在 call_next 之後取得的 Response，即是我們要攔截與修改的時機。

2. 為什麼要在 Middleware 改 Response？

需求 直接在路由處理函式完成 使用 Middleware 完成
統一回傳結構 每個路由都要寫一次，易遺漏 中介層一次設定，所有路由自動套用
加入安全 Header（如 CSP、HSTS） 需要在每個 endpoint 重複 中介層一次配置，保持一致
壓縮或加密回傳內容 需要在每個 endpoint 實作 中介層一次處理，降低耦合度
監控回應時間 / 大小 分散在多處，難以統計 中介層集中收集，便於上報

3. 基本範例：統一回傳格式

以下示範一個最簡單的 Middleware，將所有回傳的 JSON 包裝成 { "code": 0, "message": "success", "data": <原始內容> }。如果回傳不是 JSON（例如 HTML），則直接返回原始內容，避免破壞非 JSON 回應。

# middleware/response_wrapper.py
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
import json

app = FastAPI()

@app.middleware("http")
async def response_wrapper(request: Request, call_next):
    # 前置階段：可以在此記錄請求資訊
    response: Response = await call_next(request)   # 呼叫下一層

    # 後置階段：取得原始回應內容
    if response.headers.get("content-type") == "application/json":
        # 讀取原始 body（需要先轉成 bytes）
        body = b"".join([chunk async for chunk in response.body_iterator])
        # 將 bytes 轉成 dict
        original_data = json.loads(body)
        # 包裝成統一格式
        wrapped = {
            "code": 0,
            "message": "success",
            "data": original_data
        }
        return JSONResponse(content=wrapped, status_code=response.status_code)
    # 非 JSON 回應直接回傳
    return response

重點：

  • response.body_iterator 只能消費一次，若要再次使用原始內容，必須先把它讀出並重新建立 Response。
  • 為了避免破壞非 JSON 回應，我們先檢查 content-type。

4. 範例二：加入自訂 Header（安全性、跨域）

有時候後端需要在每個回應加上 安全 Header（如 X-Content-Type-Options: nosniff、X-Frame-Options: DENY），或是加入追蹤用的 X-Request-ID。下面的 Middleware 示範如何一次性完成。

# middleware/add_headers.py
from fastapi import FastAPI, Request, Response
import uuid

app = FastAPI()

@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    # 產生唯一的 request id，供日誌與追蹤使用
    request_id = str(uuid.uuid4())
    response: Response = await call_next(request)

    # 加入自訂 Header
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
    return response

技巧：若你使用 uvicorn 的 --log-level debug，可以在日誌中直接印出 X-Request-ID，方便追蹤單筆請求的全程。

5. 範例三：回應壓縮（GZIP）

在大量資料傳輸時，GZIP 壓縮 能顯著降低網路流量。Starlette 已提供 GZIPResponse，但若想在所有回應自動套用，可自行寫一個 Middleware。

# middleware/gzip_response.py
from fastapi import FastAPI, Request, Response
from starlette.responses import GZIPResponse

app = FastAPI()

@app.middleware("http")
async def gzip_middleware(request: Request, call_next):
    response: Response = await call_next(request)

    # 只對可壓縮的 MIME 類型做 GZIP
    compressible = (
        "application/json",
        "text/html",
        "text/plain",
        "text/css",
        "application/javascript",
    )
    content_type = response.headers.get("content-type", "").split(";")[0]

    if content_type in compressible and "gzip" not in request.headers.get("accept-encoding", ""):
        # 重新包裝為 GZIPResponse
        body = b"".join([chunk async for chunk in response.body_iterator])
        return GZIPResponse(content=body, status_code=response.status_code, headers=response.headers)
    return response

注意：

  • 只在客戶端 Accept-Encoding 包含 gzip 時才壓縮，避免不支援的瀏覽器收到無法解壓的回應。
  • 讀取 response.body_iterator 後，需要重新建立 Response（此例使用 GZIPResponse），否則原始 body 會被消耗掉。

6. 範例四：統一錯誤回傳格式

FastAPI 的例外處理機制（Exception Handlers）可以與 Middleware 結合，讓所有未捕捉的例外都走同一個回傳格式。

# middleware/error_formatter.py
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

@app.exception_handler(StarletteHTTPException)
@app.exception_handler(RequestValidationError)
async def http_exception_handler(request: Request, exc):
    # 這裡僅示範 HTTPException 與 ValidationError，其他自訂例外可自行加入
    return JSONResponse(
        status_code=getattr(exc, "status_code", 500),
        content={"code": getattr(exc, "status_code", 500), "message": str(exc), "data": None},
    )

@app.middleware("http")
async def error_wrapper(request: Request, call_next):
    try:
        response = await call_next(request)
        return response
    except Exception as exc:  # 捕捉未處理的例外
        return JSONResponse(
            status_code=500,
            content={"code": 500, "message": "Internal Server Error", "data": None},
        )

重點：

  • 先使用 Exception Handler 處理已知例外（如 404、422），再在 Middleware 捕捉未知例外，確保 所有 回應都有統一結構。

常見陷阱與最佳實踐

陷阱 說明 解決方式
Body 被消耗 response.body_iterator 只能讀一次，若直接 return response 之後再讀會得到空內容。 先把 body 讀進變數，然後使用 Response、JSONResponse 或 GZIPResponse 重新建立回應。
非同步與同步混用 在 async Middleware 中使用同步的 I/O（如 json.dumps）不會破壞執行，但大量同步操作會阻塞事件迴圈。 儘量使用非同步的函式（如 orjson 的 orjson.dumps）或把重 CPU 工作搬到背景執行緒。
重寫 Header 時遺失原有 Header 直接建立新 Response 時，若忘記把原始 response.headers 複製過去，會失去如 Set-Cookie 等重要資訊。 在建立新 Response 時，使用 headers=response.headers 參數或手動 copy。
對所有 MIME 類型做壓縮 壓縮圖片、影片等已壓縮檔案會浪費 CPU，且有時會增加檔案大小。 只對文字類型（JSON、HTML、CSS、JS）做 GZIP，並檢查 Accept-Encoding。
過度堆疊 Middleware 每個 Middleware 都會產生一次 await call_next(request)，過多的 Middleware 會增加請求延遲。 只保留必要的 Middleware，將相同功能合併（如同時加入 Header 與 Log）。
例外未被捕捉 在 Middleware 中直接 await call_next，若下游路由拋出例外且沒有全域 Exception Handler，會直接返回 500，且無法統一格式。 在 Middleware 包裝 try/except，或同時使用 app.exception_handler。

最佳實踐

  1. 保持 Middleware 輕量：僅處理與「跨請求」相關的事務，如 Header、統一回傳、壓縮、日誌。
  2. 使用 starlette.datastructures.MutableHeaders：若要在回應階段修改 Header，直接操作 response.headers 更安全。
  3. 測試：利用 TestClient 撰寫單元測試，確認每個 Middleware 在不同回應類型（JSON、HTML、Streaming）下的行為。
  4. 設定開關：在 settings.py 中提供 ENABLE_GZIP = True、ENABLE_RESPONSE_WRAPPER = False 等開關，讓不同環境（開發、測試、正式）可以靈活調整。
  5. 文件化：在專案的 README 或 Wiki 中說明每個 Middleware 的目的與使用方式，方便新加入的開發者快速了解。

實際應用場景

場景 需求 建議的 Middleware 組合
微服務 API Gateway 統一回傳結構、加入追蹤 Header、壓縮回傳、統一錯誤格式 response_wrapper + add_security_headers + gzip_middleware + error_wrapper
資料分析平台（大量 JSON） 減少網路流量、提供 X-Request-ID、記錄回應時間 gzip_middleware + add_security_headers + log_response_time（自訂）
前端 SPA 後端 所有回傳皆為 JSON、需要 Cache-Control、跨域安全 Header response_wrapper + add_cors_and_cache_headers（自訂）
內部管理系統 需要在錯誤時回傳友善訊息、同時保留原始例外資訊於日誌 error_wrapper + exception_handler + log_exception_middleware

範例：在 API Gateway 中使用多個 Middleware 

# main.py
from fastapi import FastAPI
from middleware.response_wrapper import response_wrapper
from middleware.add_headers import add_security_headers
from middleware.gzip_response import gzip_middleware
from middleware.error_formatter import error_wrapper

app = FastAPI()

# 依序註冊，順序會影響最終結果
app.middleware("http")(add_security_headers)   # 先加入 Header
app.middleware("http")(gzip_middleware)        # 再壓縮（已包含 Header）
app.middleware("http")(response_wrapper)      # 再統一回傳格式（會把 GZIP 後的內容包裝）
app.middleware("http")(error_wrapper)         # 最後捕捉未處理例外

# 以下是一個測試用的路由
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id, "detail": "這是一筆測試資料"}

總結

  • Middleware 是 FastAPI 中處理 Response 攔截與修改 的關鍵切入點，讓開發者可以在單一位置完成 統一回傳格式、加密/壓縮、加入 Header、錯誤統一化 等跨路由的需求。
  • 實作時須留意 body 被消耗、Header 複製、非同步效能 等常見陷阱，並以 輕量、可組合、可開關 為設計原則。
  • 透過範例可以看到，僅需幾行程式碼即可為整個 API 注入安全性、效能與可觀測性，提升專案的 可維護性 與 使用者體驗。

掌握了這些技巧後，你就能在 FastAPI 專案中靈活地控制每一次的回應，讓 API 更加 一致、可靠且易於監控。祝開發順利！