FastAPI 基礎概念:Starlette(ASGI)與 Pydantic
簡介
在 FastAPI 這個高效能的 Web 框架背後,兩個核心技術決定了它的速度與易用性:Starlette(一個基於 ASGI 的底層框架)與 Pydantic(資料驗證與序列化的利器)。了解這兩者的運作機制,對於寫出 安全、可維護且效能卓越 的 API 至關重要。
本篇文章將從概念層面解說 Starlette 與 ASGI 的關係、Pydantic 的資料模型原理,並透過實作範例展示它們在 FastAPI 中的實際應用。即使你是剛接觸 FastAPI 的新手,只要跟著步驟操作,也能快速掌握這些底層工具的使用方式,為日後開發更複雜的服務打下堅實基礎。
核心概念
1. ASGI 與 Starlette
1.1 為什麼是 ASGI?
傳統的 WSGI(Web Server Gateway Interface)只能同步處理請求,當請求需要 I/O(如資料庫、外部 API)時,整個工作執行緒會被阻塞。
ASGI(Asynchronous Server Gateway Interface) 則是為了支援 非同步(async)程式設計而誕生,允許單一執行緒同時處理多個請求,極大提升效能與資源利用率。
1.2 Starlette:ASGI 的「鋼鐵底層」
Starlette 是一個輕量級的 ASGI 框架,提供:
- 路由(routing)與 URL 解析
- 中介層(middleware)機制
- WebSocket、背景任務(background tasks)支援
- 靜態檔案服務
FastAPI 把 Starlette 當作 底層引擎,把路由、請求/回應處理交給它,同時在上層加入自動產生 OpenAPI 文件、依賴注入等功能。
2. Pydantic:型別安全的資料模型
2.1 資料驗證的痛點
在傳統的 Flask/Django 中,開發者往往自行撰寫表單或序列化程式碼,容易遺漏欄位檢查、型別轉換或錯誤訊息統一的問題。
2.2 Pydantic 的解法
Pydantic 以 Python 型別提示(type hints) 為基礎,利用 dataclasses 的概念自動完成:
- 欄位驗證(如
int,EmailStr,conlist等) - 資料轉換(字串 → datetime、字串 → Enum)
- 錯誤回傳(FastAPI 會直接把驗證失敗的訊息轉成 422 回應)
2.3 與 FastAPI 的結合
當你在 FastAPI 的路由函式中宣告參數型別為 Pydantic model 時,框架會自動:
- 從請求的 JSON body、query parameters、path parameters 中取值
- 用 Pydantic 產生模型實例,完成驗證與轉換
- 若驗證失敗,回傳結構化的錯誤訊息
程式碼範例
以下範例展示 Starlette、ASGI 與 Pydantic 在 FastAPI 中的實作細節。每段程式碼均附有說明,方便理解背後的運作原理。
範例 1:最小化的 ASGI 應用(直接使用 Starlette)
# app_starlette.py
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
async def homepage(request):
# 這是一個非同步的處理函式,符合 ASGI 規範
return JSONResponse({"message": "Hello from Starlette!"})
routes = [
Route("/", endpoint=homepage, methods=["GET"]),
]
app = Starlette(debug=True, routes=routes)
重點:
async def讓函式在 事件迴圈 中非同步執行。- Starlette 直接回傳
JSONResponse,不需要額外的序列化程式碼。
範例 2:在 FastAPI 中使用 Pydantic 進行請求驗證
# app_fastapi.py
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, conint
app = FastAPI(title="User API")
class UserCreate(BaseModel):
"""建立使用者時的資料模型"""
username: str # 必填字串
email: EmailStr # 自動驗證 Email 格式
age: conint(gt=0, lt=150) # 年齡必須在 1~149 之間
@app.post("/users/", response_model=UserCreate)
async def create_user(user: UserCreate):
"""
FastAPI 會自動:
1. 從 request body 解析 JSON
2. 用 Pydantic 產生 UserCreate 實例
3. 若驗證失敗回傳 422
"""
# 這裡可以直接使用已驗證的 user 物件
return user
說明:
BaseModel讓我們只需要宣告欄位型別,即可得到完整的驗證與序列化功能。response_model會在回傳前再一次使用 Pydantic 產生 輸出模型,確保回應資料的一致性。
範例 3:自訂 Starlette 中介層(Middleware)結合 FastAPI
# middleware_example.py
import time
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import FastAPI, Request
class TimingMiddleware(BaseHTTPMiddleware):
"""計算每筆請求的執行時間,示範如何在 Starlette 中加入自訂中介層"""
async def dispatch(self, request: Request, call_next):
start = time.time()
response = await call_next(request) # 呼叫下游的 FastAPI 處理函式
process_time = (time.time() - start) * 1000
response.headers["X-Process-Time-ms"] = str(round(process_time, 2))
return response
app = FastAPI()
app.add_middleware(TimingMiddleware)
@app.get("/ping")
async def ping():
return {"msg": "pong"}
關鍵:
BaseHTTPMiddleware來自 Starlette,FastAPI 直接支援加入。call_next仍然是 ASGI 呼叫,保持非同步特性。
範例 4:使用 Pydantic 的自訂驗證器(Validator)
# validator_example.py
from fastapi import FastAPI
from pydantic import BaseModel, validator
from datetime import datetime
app = FastAPI()
class Event(BaseModel):
name: str
start_time: datetime
end_time: datetime
@validator("end_time")
def check_time_order(cls, v, values):
"""確保結束時間晚於開始時間"""
if "start_time" in values and v <= values["start_time"]:
raise ValueError("end_time 必須晚於 start_time")
return v
@app.post("/events/")
async def create_event(event: Event):
return {"status": "created", "event": event}
說明:
@validator讓我們在模型層面加入 業務規則,避免在路由函式內重複檢查。
範例 5:非同步背景任務(Background Tasks)— 透過 Starlette 的功能
# background_task.py
from fastapi import FastAPI, BackgroundTasks
import asyncio
app = FastAPI()
async def write_log(message: str):
# 假裝寫入慢速的 I/O(例如寫檔或發送 email)
await asyncio.sleep(1)
print(f"[LOG] {message}")
@app.post("/notify/")
async def send_notification(email: str, background_tasks: BackgroundTasks):
# 立即回應客戶端,將寫入 log 的工作交給背景執行
background_tasks.add_task(write_log, f"Sent notification to {email}")
return {"msg": "notification scheduled"}
重點:
BackgroundTasks內部實作基於 Starlette 的非同步機制,讓我們在不阻塞主請求的情況下完成 I/O 任務。
常見陷阱與最佳實踐
| 陷阱 | 可能的後果 | 建議的解決方案 |
|---|---|---|
在路由函式內使用阻塞 I/O(如 requests.get()) |
事件迴圈被卡住,導致所有請求變慢 | 改用 非同步 HTTP 客戶端(如 httpx.AsyncClient)或將阻塞程式碼搬到執行緒池 (run_in_threadpool) |
忘記在 Pydantic model 中使用 Field(..., ...) 設定必填 |
欄位被視為可選,導致資料不完整卻不會拋錯 | 明確使用 ... 表示必填,或在 Config 中設定 extra = "forbid" 防止多餘欄位 |
在自訂 Middleware 中忘記呼叫 await call_next(request) |
請求永遠不會到達下游路由,回應超時 | 確保 dispatch 方法最後返回 await call_next(request) 的結果 |
使用 list 作為 Pydantic 欄位的預設值 |
每次建立模型時共用同一個 list,造成資料污染 | 使用 default_factory=list 或 Field(default_factory=list) |
| 在 FastAPI 中混用同步與非同步函式(尤其在同一路由) | 可能產生不必要的執行緒切換,降低效能 | 盡量保持一致:若路由使用 async def,內部呼叫的函式也應該是非同步的;若必須使用同步函式,考慮使用 run_in_threadpool 包裝 |
最佳實踐
- 型別提示 + Pydantic:所有外部輸入(body、query、path、header)都用 Pydantic model 包裝,讓驗證在最早階段完成。
- 全域例外處理:利用 FastAPI 的
exception_handler統一回傳錯誤格式,提升前端除錯效率。 - 分層設計:將路由、服務層(service)與資料層(repository)分離,保持 FastAPI 只負責 協調,而非業務邏輯。
- 使用測試客戶端:
from fastapi.testclient import TestClient搭配 pytest,驗證 Pydantic model 與路由行為。 - 效能監控:加入 Starlette 的中介層紀錄每筆請求的執行時間,或使用 Prometheus、Grafana 觀測 API 延遲。
實際應用場景
| 場景 | 為何選擇 FastAPI + Starlette + Pydantic |
|---|---|
| 微服務 API | ASGI 的非同步特性讓高併發請求不會被阻塞,且自動產生的 OpenAPI 文件降低跨團隊溝通成本。 |
| 資料驗證密集的表單系統 | Pydantic 能在模型層即完成複雜驗證(例如日期範圍、正則表達式、嵌套結構),減少手寫驗證程式碼。 |
| 即時推播或 WebSocket 應用 | Starlette 原生支援 WebSocket,FastAPI 只需要在路由上加上 WebSocket 類型即可,實作簡潔。 |
| 背景任務與排程 | 透過 BackgroundTasks 或與 Celery/RQ 結合,能在不阻塞主請求的情況下完成長時間任務。 |
| 資料科學模型服務 | 使用非同步 HTTP 客戶端呼叫外部機器學習服務,或直接在 FastAPI 中載入模型(如 TensorFlow、PyTorch),利用 ASGI 的效能處理大量推論請求。 |
總結
- Starlette 為 FastAPI 提供了 ASGI 的基礎設施,使框架具備非同步、WebSocket、背景任務等現代 Web API 所必需的功能。
- Pydantic 則透過 型別提示 與 自動驗證,把資料安全與序列化的工作在模型層完成,讓開發者可以把注意力集中在業務邏輯上。
- 兩者的結合讓 FastAPI 成為 「開發速度快、執行效能高、文件完整」 的首選框架。
只要掌握了 ASGI 的概念、熟悉 Starlette 的中介層與路由機制,同時善用 Pydantic 的驗證功能,你就能在專案中建立 高品質、易維護且具備彈性 的 API,無論是小型原型還是大型微服務,都能應付自如。祝開發順利,期待你在 FastAPI 的世界裡寫出更多精彩的程式碼!