目錄
06-1. 依賴注入基礎
⏱️ 閱讀時間: 15 分鐘 🎯 難度: ⭐⭐ (中級)
🤔 一句話解釋
依賴注入(DI)是一種設計模式,讓函數的依賴由外部提供,而不是自己創建,提高程式碼的可測試性和可維護性。
🔄 沒有 DI vs 有 DI
# ❌ 沒有依賴注入
def get_users():
db = Database("postgresql://localhost/db") # 硬編碼
return db.query("SELECT * FROM users")
# ✅ 有依賴注入
def get_users(db: Database): # 依賴由外部提供
return db.query("SELECT * FROM users")沒有 DI:
┌────────────┐ ┌──────────────┐
│ get_users │────▶│ Database │ 內部建立
└────────────┘ └──────────────┘
緊密耦合,難以測試
有 DI:
┌────────────┐ ┌──────────────┐
│ 外部 │────▶│ Database │
└────────────┘ └──────────────┘
│ │
└─────────────────▶│
┌────────────┐
│ get_users │ 接收依賴
└────────────┘
鬆散耦合,易於測試🎯 依賴注入的好處
| 好處 | 說明 |
|---|---|
| 可測試性 | 可以注入 mock 物件進行測試 |
| 可維護性 | 依賴關係清晰,易於修改 |
| 可重用性 | 相同的依賴可以在多處使用 |
| 鬆散耦合 | 組件之間不直接相依 |
📦 FastAPI 依賴注入
基本用法
from fastapi import FastAPI, Depends
app = FastAPI()
# 定義依賴函數
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
# 使用依賴
@app.get("/items")
async def read_items(commons: dict = Depends(common_parameters)):
return {"commons": commons}
@app.get("/users")
async def read_users(commons: dict = Depends(common_parameters)):
return {"commons": commons}依賴的執行順序
客戶端請求
│
▼
┌─────────────────┐
│ 解析依賴樹 │
└─────────────────┘
│
▼
┌─────────────────┐
│ 執行依賴函數 │ ← common_parameters()
└─────────────────┘
│
▼
┌─────────────────┐
│ 執行端點函數 │ ← read_items()
└─────────────────┘
│
▼
返回回應🔧 依賴類型
函數依賴
from fastapi import FastAPI, Depends, Query
app = FastAPI()
def pagination(
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100)
):
"""分頁參數依賴"""
return {
"skip": (page - 1) * page_size,
"limit": page_size
}
@app.get("/items")
async def get_items(pagination: dict = Depends(pagination)):
# pagination = {"skip": 0, "limit": 20}
return {"pagination": pagination}類別依賴
from fastapi import FastAPI, Depends, Query
app = FastAPI()
class Pagination:
"""分頁類別依賴"""
def __init__(
self,
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100)
):
self.page = page
self.page_size = page_size
self.skip = (page - 1) * page_size
self.limit = page_size
@app.get("/items")
async def get_items(pagination: Pagination = Depends(Pagination)):
return {
"page": pagination.page,
"skip": pagination.skip,
"limit": pagination.limit
}
# 簡化寫法(Depends() 可以省略參數)
@app.get("/users")
async def get_users(pagination: Pagination = Depends()):
return {"page": pagination.page}可呼叫類別
from fastapi import FastAPI, Depends
app = FastAPI()
class DatabaseChecker:
"""可呼叫的類別依賴"""
def __init__(self, db_name: str):
self.db_name = db_name
def __call__(self):
# 檢查資料庫連線
return {"database": self.db_name, "status": "connected"}
# 建立依賴實例
check_primary_db = DatabaseChecker("primary")
check_replica_db = DatabaseChecker("replica")
@app.get("/primary")
async def primary_endpoint(db_status: dict = Depends(check_primary_db)):
return db_status
@app.get("/replica")
async def replica_endpoint(db_status: dict = Depends(check_replica_db)):
return db_status📊 非同步依賴
from fastapi import FastAPI, Depends
import httpx
app = FastAPI()
async def get_external_data():
"""非同步依賴"""
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com/data")
return response.json()
@app.get("/items")
async def get_items(external_data: dict = Depends(get_external_data)):
return {"external": external_data}🔄 巢狀依賴
依賴之間的依賴
from fastapi import FastAPI, Depends, Query
app = FastAPI()
def query_extractor(q: str = Query(None)):
"""提取查詢參數"""
return q
def query_or_default(q: str = Depends(query_extractor)):
"""依賴於 query_extractor"""
if q:
return q
return "default"
@app.get("/items")
async def get_items(query: str = Depends(query_or_default)):
return {"query": query}多層依賴
from fastapi import FastAPI, Depends, Header, HTTPException
app = FastAPI()
async def verify_token(x_token: str = Header()):
"""驗證 Token"""
if x_token != "secret-token":
raise HTTPException(status_code=401, detail="Invalid token")
return x_token
async def get_current_user(token: str = Depends(verify_token)):
"""取得當前使用者(依賴於 verify_token)"""
# 根據 token 查詢使用者
return {"user_id": 1, "username": "john"}
async def get_user_permissions(
user: dict = Depends(get_current_user)
):
"""取得使用者權限(依賴於 get_current_user)"""
return {"user": user, "permissions": ["read", "write"]}
@app.get("/admin")
async def admin_endpoint(
permissions: dict = Depends(get_user_permissions)
):
return permissions依賴樹:
admin_endpoint
│
└── get_user_permissions
│
└── get_current_user
│
└── verify_token⚙️ 路徑操作裝飾器中的依賴
from fastapi import FastAPI, Depends, Header, HTTPException
app = FastAPI()
async def verify_key(x_key: str = Header()):
if x_key != "valid-key":
raise HTTPException(status_code=403, detail="Invalid key")
return x_key
# 依賴在裝飾器中,不注入到函數參數
@app.get("/items", dependencies=[Depends(verify_key)])
async def get_items():
return {"items": []}
@app.get("/users", dependencies=[Depends(verify_key)])
async def get_users():
return {"users": []}🌐 全域依賴
from fastapi import FastAPI, Depends, Header, HTTPException
async def verify_api_key(x_api_key: str = Header()):
if x_api_key != "my-api-key":
raise HTTPException(status_code=403)
# 所有端點都會執行這個依賴
app = FastAPI(dependencies=[Depends(verify_api_key)])
@app.get("/items")
async def get_items():
return {"items": []}
@app.get("/users")
async def get_users():
return {"users": []}📝 實戰範例:認證依賴
from fastapi import FastAPI, Depends, HTTPException, Header
from pydantic import BaseModel
from typing import Optional
app = FastAPI()
class User(BaseModel):
id: int
username: str
email: str
is_active: bool = True
is_admin: bool = False
# 模擬資料庫
fake_users = {
"token_user1": User(id=1, username="john", email="john@example.com"),
"token_admin": User(
id=2, username="admin", email="admin@example.com", is_admin=True
),
}
async def get_token(authorization: str = Header()):
"""提取 Token"""
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Invalid authorization header")
return authorization.replace("Bearer ", "")
async def get_current_user(token: str = Depends(get_token)) -> User:
"""取得當前使用者"""
user = fake_users.get(token)
if not user:
raise HTTPException(status_code=401, detail="Invalid token")
return user
async def get_current_active_user(
user: User = Depends(get_current_user)
) -> User:
"""確保使用者是啟用狀態"""
if not user.is_active:
raise HTTPException(status_code=403, detail="Inactive user")
return user
async def get_admin_user(
user: User = Depends(get_current_active_user)
) -> User:
"""確保使用者是管理員"""
if not user.is_admin:
raise HTTPException(status_code=403, detail="Admin required")
return user
# 一般使用者端點
@app.get("/users/me")
async def read_me(user: User = Depends(get_current_active_user)):
return user
# 管理員端點
@app.get("/admin/users")
async def read_all_users(admin: User = Depends(get_admin_user)):
return list(fake_users.values())✅ 重點總結
依賴類型
| 類型 | 說明 |
|---|---|
| 函數 | 簡單依賴 |
| 類別 | 需要 __init__ 參數 |
| 可呼叫 | 實作 __call__ 方法 |
使用方式
| 方式 | 用途 |
|---|---|
Depends(func) | 函數參數中使用 |
dependencies=[...] | 裝飾器中使用 |
app = FastAPI(dependencies=[...]) | 全域使用 |
最佳實踐
- 保持依賴函數單一職責
- 使用類型提示提高可讀性
- 善用巢狀依賴組織邏輯
- 考慮依賴的效能影響
🎤 面試這樣答
Q: 什麼是依賴注入?FastAPI 如何實現?
答案:
依賴注入是一種設計模式,讓函數的依賴由外部提供,而非自己創建。
FastAPI 實現方式:
from fastapi import Depends def get_db(): db = Database() try: yield db finally: db.close() @app.get("/users") async def get_users(db: Database = Depends(get_db)): return db.get_all_users()好處:
- 可測試性:可以注入 mock
- 可維護性:依賴關係清晰
- 程式碼重用:相同依賴多處使用
下一篇: 06-2. yield 依賴
最後更新:2025-12-18