FastAPI 依賴注入 (Dependency Injection)
依賴注入 (Dependency Injection, DI) 是 FastAPI 設計的核心,也是它如此靈活、強大的原因。
即使你沒聽過「依賴注入」這個詞,你也沒必要害怕。這只是一個花俏的名詞,實際上它的概念非常簡單。
什麼是依賴注入?
簡單來說,「依賴注入」就是讓你的程式碼(例如 API 路徑操作函數)宣告它需要什麼東西(依賴項),然後由系統(FastAPI)負責在執行時把這些東西「注入」進來。
這些「東西」可能是:
- 資料庫連線
- 目前登入的使用者
- 權限驗證邏輯
- 共用的查詢參數處理
建立第一個依賴項
讓我們來看一個簡單的例子。假設我們有多個路徑都需要處理分頁查詢參數 (skip 和 limit)。
如果不使用 DI,你可能要在每個函數都寫一次 skip: int = 0, limit: int = 10。
使用 DI,我們可以這樣做:
1. 建立一個依賴函數
from fastapi import FastAPI, Depends
app = FastAPI()
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
這個 common_parameters 函數看起來就像一般的 Path Operation Function,它接收參數,然後回傳一個 dict。
2. 在路徑操作中使用 Depends
接下來,我們在 API 路由中使用 Depends 來宣告這個依賴:
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
return commons
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
return commons
發生了什麼事?
在現代 FastAPI 開發中,官方推薦使用 typing.Annotated 來宣告依賴項,這能讓你的程式碼更具可讀性並減少重複:
from typing import Annotated
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
return commons
當你存取 /items/?q=foo&skip=20&limit=50 時:
- FastAPI 看到了
read_items需要參數commons。 commons是一個依賴項,由Depends(common_parameters)定義。- FastAPI 呼叫
common_parameters函數,並將 URL 中的參數 (q,skip,limit) 傳給它。 common_parameters執行完畢,回傳{"q": "foo", "skip": 20, "limit": 50}。- FastAPI 將這個回傳值賦值給
read_items的commons參數。 - 最後執行
read_items。
型別提示與相容性
在上面的例子中,read_items 裡的 commons 型別標註為 dict。雖然這在功能上可行,但編輯器無法得知 dict 裡面有哪些 key。
為了更好的編輯器支援,你可以使用 類別 (Class) 作為依賴項:
from typing import Annotated
from fastapi import Depends
class CommonQueryParams:
def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):
# 現在你可以享受自動補全,例如 commons.q, commons.limit
return commons
這樣一來,FastAPI 會自動實例化該類別,而你的 IDE 也能提供完整的屬性補全與型別檢查。
Sync 與 Async 依賴項
FastAPI 設計得非常靈活,它不限制你的依賴項必須是同步 (def) 或非同步 (async def)。你可以根據需求自由選擇,甚至在同一個路由中混用兩者。
靈活混用
你可以在 async def 的路由中使用 def 的依賴項,也可以在 def 的路由中使用 async def 的依賴項。FastAPI 會自動幫你處理好正確的呼叫方式。
def get_db():
# 同步依賴項(例如連線 MySQL 或 SQLite)
return "DB Connection"
async def get_current_user(token: str):
# 非同步依賴項(例如需要 await 的 API 請求)
return {"user": "mike"}
@app.get("/items/")
async def read_items(db = Depends(get_db), user = Depends(get_current_user)):
# 這裡可以同時使用同步與非同步的依賴項
return {"db": db, "user": user}
該如何選擇?
- 使用
async def:如果你的依賴項內部需要執行非同步操作(例如使用httpx.AsyncClient存取外部 API,或是使用異步的資料庫驅動程式元件),請務必使用async def。 - 使用
def:如果依賴項純粹是邏輯運算,或者你使用的是傳統的同步資料庫驅動(如 SQLAlchemy 或 Psycopg2 的同步模式),使用一般的def即可。
def 依賴項,FastAPI 仍會在外部執行緒池 (threadpool) 中執行它,以確保不會阻塞主事件迴圈的運行。在路徑裝飾器中使用依賴項
有時候,你只需要依賴項被「執行」,而不需要在你的路由函數中使用它的回傳值。
例如:
- 檢查一個自定義的 Header。
- 驗證使用者的權限,但不需要完整的使用者資料。
在這種情況下,你可以使用路徑裝飾器(例如 @app.get)中的 dependencies 參數:
from fastapi import Depends, FastAPI, Header, HTTPException
app = FastAPI()
async def verify_token(x_token: Annotated[str, Header()]):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def verify_key(x_key: Annotated[str, Header()]):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
return [{"item": "Portal Gun"}, {"item": "Plumbus"}]
與一般 Depends 的差異
- 參數宣告:
dependencies接收一個list,你可以一次放入多個依賴項。 - 回傳值:這些依賴項的回傳值會被丟棄,不會注入到你的路由函數中。
- 執行時機:它們與一般的
Depends一樣,會由左至右依序執行。
當你有一系列通用的權限校驗或安全性檢查,且不想讓你的路由函數參數列表變得又臭又長時,這是一個非常乾淨的作法。
為何要使用依賴注入?
- 程式碼重用 (Code Reuse):把重複的邏輯 (如分頁、驗證) 抽離出來,寫一次,到處用。
- 關注點分離 (Separation of Concerns):路徑操作函數只需專注於處理請求,不用管底層的資料庫連線或權限檢查細節。
- 測試容易:在寫單元測試時,你可以輕鬆地用 Mock 物件替換掉真實的依賴項 (例如把真實資料庫換成測試資料庫)。
總結
- 定義依賴:寫一個函數,宣告需要的參數 (Query, Path, Body 等)。
- 使用依賴:在路由函數參數中使用
Depends(DependencyFunction)。 - FastAPI 負責解析參數 -> 呼叫依賴函數 -> 取得結果 -> 注入給路由函數。