FastAPI 請求生命週期（Request Lifecycle）

簡介

在 Web 框架中，請求生命週期是指從客戶端發送 HTTP 請求，到伺服器回傳回應的完整過程。了解這個流程不僅能幫助我們寫出更可維護、效能更佳的 API，也能在除錯、日誌、驗證、授權等情境下，正確切入適當的階段進行處理。

FastAPI 以 Starlette 為底層，採用 ASGI（Asynchronous Server Gateway Interface）規範，使得請求的每一步都可以被自訂的中介層（middleware）或依賴注入（dependency）所攔截。掌握這些機制，對於從「Hello World」到「企業級微服務」的遷移尤為重要。

本篇文章將深入說明 FastAPI 處理請求的完整流程，並提供實作範例、常見陷阱與最佳實踐，讓你在開發過程中能夠 快速定位問題、優化效能、提升安全性。

核心概念

1. ASGI 應用程式與事件迴圈

FastAPI 其實是一個 ASGI 應用程式，在啟動時會被 ASGI 伺服器（如 Uvicorn、Hypercorn）載入，並在單一事件迴圈（event loop）中執行非同步 I/O。這意味著每一次請求都會在同一個事件迴圈內被「await」處理，而不會阻塞其他請求。

重點：若在路由函式中使用阻塞式 I/O（如 time.sleep()），會阻塞整個事件迴圈，造成所有連線卡住。請改用 await asyncio.sleep() 或將阻塞程式碼搬到執行緒池（run_in_threadpool）。

2. 中介層（Middleware）

中介層是 先於路由處理、後於路由回傳 的函式，能夠在請求與回應之間加入自訂邏輯。FastAPI 允許多個中介層以堆疊方式運作，最早加入的最先執行（類似「洋蔥模型」）。

from fastapi import FastAPI, Request
from starlette.responses import Response
import time

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    """計算每個請求的處理時間，並在回應標頭中回傳"""
    start_time = time.time()
    response: Response = await call_next(request)   # 呼叫下一層（可能是其他 middleware 或路由）
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

說明：call_next 代表「把請求往下傳遞」的函式，必須 await，否則請求會卡住。

3. 依賴注入（Dependency Injection）

FastAPI 把依賴視為「在請求生命週期的特定階段」自動執行的函式。依賴可以設定 作用範圍（scope），最常見的是 request（每個請求一次）與 singleton（應用程式啟動時一次）。

from fastapi import Depends, Header, HTTPException

async def verify_token(x_token: str = Header(...)):
    """驗證自訂 Header 中的 token，若失敗則拋出 401"""
    if x_token != "secret-token":
        raise HTTPException(status_code=401, detail="Invalid X-Token")
    return x_token

@app.get("/items/", dependencies=[Depends(verify_token)])
async def read_items():
    return [{"item_id": "Foo"}, {"item_id": "Bar"}]

重點：依賴會在 路由函式執行前 被解析，且可以返回值供路由使用（或僅作為 side‑effect）。

4. 路由函式（Endpoint）

路由函式是 請求的核心，接收經過依賴、驗證、解析後的參數，執行業務邏輯，最後返回 Response（或任意可序列化的物件）。FastAPI 會根據函式的型別提示自動產生 OpenAPI 文件與資料驗證。

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    tags: list[str] = []

@app.post("/items/")
async def create_item(item: Item):
    # 假設此處寫入資料庫（使用 async ORM）
    return {"msg": "Item created", "item": item}

5. 回應處理與例外捕捉

在路由函式返回後，FastAPI 會將結果轉換成 Starlette Response。如果在任意階段拋出例外，FastAPI 會走 exception handlers（預設或自訂）產生統一的錯誤回應。

from fastapi import Request, status
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail, "path": request.url.path},
    )

程式碼範例

以下提供 五個實務上常用 的範例，示範如何在請求生命週期的不同階段加入功能。

範例 1️⃣：全局請求日誌（Middleware）

import json
import logging
from fastapi import FastAPI, Request
from starlette.responses import Response

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

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """將每筆請求的 method、url、client IP、body 記錄到日誌"""
    body = await request.body()
    logger.info(
        json.dumps({
            "method": request.method,
            "url": str(request.url),
            "client": request.client.host,
            "body": body.decode() if body else None,
        })
    )
    response: Response = await call_next(request)
    return response

說明：await request.body() 只能讀一次，若後續路由仍需要讀取請求體，請使用 request = Request(request.scope, receive=await request.receive) 重新包裝，或改用 request.stream()。

範例 2️⃣：資料庫連線（依賴，作用域 = `request`）

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from fastapi import Depends

DATABASE_URL = "postgresql+asyncpg://user:pwd@localhost/db"

engine = create_async_engine(DATABASE_URL, echo=False)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db() -> AsyncSession:
    """每個請求建立一個 Session，請求結束自動關閉"""
    async with AsyncSessionLocal() as session:
        yield session   # 使用 yield，FastAPI 會在回傳後自動執行 finally 區塊

@app.get("/users/{user_id}")
async def read_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute("SELECT * FROM users WHERE id = :id", {"id": user_id})
    user = result.fetchone()
    return {"user": dict(user)} if user else {"detail": "User not found"}

重點：使用 async with + yield 可確保 請求結束時自動釋放連線，避免連線洩漏。

範例 3️⃣：自訂驗證 Header（Dependency）

from fastapi import Header, HTTPException, Depends

async def verify_api_key(x_api_key: str = Header(...)):
    """檢查 X-API-Key 是否符合系統設定"""
    if x_api_key != "my-super-secret":
        raise HTTPException(status_code=403, detail="Invalid API Key")
    return x_api_key

@app.get("/protected", dependencies=[Depends(verify_api_key)])
async def protected_endpoint():
    return {"msg": "You have access!"}

說明：將依賴放在 dependencies= 參數中，表示 不需要在函式內取得返回值，僅作為前置檢查即可。

範例 4️⃣：請求結束後的清理工作（Middleware + BackgroundTask）

from fastapi import BackgroundTasks

@app.post("/upload/")
async def upload_file(file: bytes, background_tasks: BackgroundTasks):
    # 假設把檔案寫入暫存目錄
    temp_path = f"/tmp/{int(time.time())}.bin"
    with open(temp_path, "wb") as f:
        f.write(file)

    # 背景任務：上傳至雲端儲存，完成後刪除本地檔案
    background_tasks.add_task(upload_to_s3_and_cleanup, temp_path)
    return {"msg": "File received, processing in background"}

async def upload_to_s3_and_cleanup(path: str):
    # 這裡使用 async S3 客戶端 (偽代碼)
    await s3_client.upload_file(path, bucket="my-bucket")
    os.remove(path)

技巧：BackgroundTasks 會在 回應送出後 執行，適合做非即時、耗時的工作（如寫入日誌、發送通知）。

範例 5️⃣：全局例外處理（Exception Handler）

from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse

@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    """捕捉未處理的例外，回傳統一格式的錯誤訊息"""
    return JSONResponse(
        status_code=500,
        content={
            "error": "Internal Server Error",
            "detail": str(exc),
            "path": request.url.path,
        },
    )

提醒：不要在此處直接 raise，否則會形成遞迴。若想保留原始例外，可在 logging 中記錄後再回傳通用訊息。

常見陷阱與最佳實踐

陷阱	說明	解決方式 / 最佳實踐
阻塞 I/O	在 async 路由中使用同步函式（如 `requests.get`）會阻塞事件迴圈。	使用 `httpx.AsyncClient` 或 `await anyio.to_thread.run_sync(sync_func)`。
多次讀取 Request Body	`await request.body()` 只能讀一次，後續再讀會得到空字串。	若需要在 middleware 與路由同時讀取，先把 body 存至 `request.state.body`，或使用 `request.stream()`。
依賴作用域錯誤	把耗時資源（如 DB 連線）設為 `singleton`，導致共享同一個連線，可能產生競爭條件。	依賴預設為 `request`，除非確定資源是 thread‑safe，才改為 `singleton`。
未捕捉例外	例外未被 handler 捕捉，會直接回傳 500，且訊息可能洩漏。	為常見例外（`HTTPException`、`ValidationError`）自訂 handler，並在 production 用統一的 `Exception` handler 隱藏細節。
過度使用全域 Middleware	把所有邏輯都寫在單一 middleware，會讓請求流程難以追蹤。	按功能分層（logging、auth、metrics），保持每個 middleware 單一職責。

最佳實踐總結

保持非同步：盡可能使用 async 函式與 async 客戶端。
依賴最小化：只在需要時才注入依賴，避免不必要的 DB 連線建立。
使用 BackgroundTask 處理非即時工作，減少回應延遲。
統一例外處理：在應用程式入口設定全域 Exception handler，確保錯誤資訊安全且易於除錯。
測試請求流程：使用 TestClient（同步）或 AsyncClient（非同步）驗證 middleware、dependency、exception 的順序與行為。

實際應用場景

場景	需求	生命週期切入點	範例實作
API 金鑰驗證	每筆請求必須檢查 `X-API-Key` 是否正確。	Dependency（`verify_api_key`）	範例 3️⃣
請求速率限制（Rate Limiting）	同一 IP 每分鐘只能呼叫 100 次。	Middleware（在 `call_next` 前檢查 Redis 計數）	`add_process_time_header` 可改寫為速率限制中介層
多租戶資料隔離	根據 JWT 中的 `tenant_id` 使用不同資料庫 schema。	Dependency（解析 JWT → 產生 `db` 連線）	`get_db` 依賴改為根據 `tenant_id` 動態建立 engine
請求追蹤（Tracing）	將每筆請求的 latency、path、status 寫入分散式追蹤系統。	Middleware + BackgroundTask（回傳後送資料）	`log_requests` + `background_tasks.add_task(send_to_otel)`
上傳大檔案的分段處理	前端分段上傳，後端在最後一步組合檔案並觸發後續工作。	Endpoint（接收分段） + BackgroundTask（組合 & 上傳）	範例 4️⃣（改寫為分段合併）

總結

FastAPI 的 請求生命週期 由底層的 ASGI 事件迴圈、全域 Middleware、依賴注入、路由處理與例外回應組成。透過合理的 中介層、依賴範圍、以及 背景任務，我們可以在每一個階段插入日誌、驗證、資源管理與清理工作，從而打造 高效、可測試、易維護 的 API。

掌握以下要點，即可在實務開發中游刃有餘：

非同步永遠是第一選項，避免阻塞事件迴圈。
Middleware 用於跨請求的橫切關注點（logging、metrics、auth）。
Dependency 以「每請求一次」的預設作用域管理資源，必要時自訂 scope。
Exception Handler 統一錯誤輸出，保護內部實作不被外洩。
BackgroundTask 處理非即時、耗時工作，提升回應速度。

只要依照上述流程與最佳實踐設計，你的 FastAPI 專案將能在 可擴展性、性能與安全性 三方面同時獲得顯著提升。祝開發順利，期待你在 FastAPI 上構建出更多優秀的服務！