在FastAPI的世界裏，依賴注入（Dependency Injection，簡稱DI）是管理代碼中資源共享和複用的核心工具。尤其在處理異步任務時，依賴注入能幫助我們優雅地處理數據庫連接、認證信息、配置參數等“資源”，避免代碼重複和耦合。本文將一步步拆解異步依賴注入的核心概念和實戰技巧，適合剛接觸FastAPI的開發者。
一、什麼是依賴注入？爲什麼重要？¶
想象你寫了一個路由函數，需要查詢數據庫獲取用戶信息。傳統方式下，你可能直接在函數里寫create_db_connection()來獲取連接，再執行查詢。但這樣做的問題是：
- 連接對象分散在各個函數中，重複創建和關閉會浪費資源；
- 如果數據庫連接邏輯變更（比如換成異步驅動），需要修改所有依賴它的函數。
依賴注入的核心思想：把“資源”（比如數據庫連接）的獲取和管理交給外部系統，函數只需聲明“我需要這個資源”，而不是自己去獲取。這樣代碼更清晰、解耦性更強，也更容易測試和維護。
二、FastAPI中的依賴注入基礎¶
FastAPI通過Depends()工具聲明依賴項，它的語法很簡單：參數 = Depends(依賴項函數)。依賴項可以是普通函數（同步）或異步函數（async def），FastAPI會自動處理調用邏輯。
1. 同步依賴注入（基礎）¶
先看一個簡單的同步例子，比如獲取數據庫連接：
from fastapi import FastAPI, Depends
# 假設這是一個同步數據庫連接函數
def get_sync_db():
db = create_sync_db_connection() # 假設這是同步創建的連接
return db
app = FastAPI()
@app.get("/users")
def read_users(db=Depends(get_sync_db)):
# 使用db查詢用戶數據（假設db是同步對象）
return db.query("SELECT * FROM users")
這裏get_sync_db是同步函數，Depends(get_sync_db)告訴FastAPI：“先執行get_sync_db()，把結果傳給db參數”。
2. 異步依賴注入（關鍵）¶
當依賴項需要異步操作（比如異步數據庫查詢、異步HTTP請求）時，只需將依賴項函數定義爲async def，FastAPI會自動用await調用它。
from fastapi import FastAPI, Depends
import asyncio
# 異步數據庫連接函數
async def get_async_db():
async with aio_db.AsyncConnection() as db: # 假設aio_db是異步數據庫驅動
yield db # 這裏用yield實現上下文管理器（可選），但也可以直接返回
# 或者更簡單地返回連接對象
async def get_async_db():
db = await aio_db.connect() # 異步連接數據庫
return db
app = FastAPI()
@app.get("/users")
async def read_users(db=Depends(get_async_db)):
# 使用await調用異步數據庫查詢
return await db.fetch("SELECT * FROM users")
關鍵點：
- 依賴項函數必須用async def定義；
- Depends(get_async_db)會自動await調用該函數，最終返回db對象（已處理完異步操作）。
三、異步依賴注入的進階技巧¶
1. 依賴項的嵌套與組合¶
複雜場景下，一個依賴項可能需要依賴另一個依賴項。例如，查詢用戶時需要先獲取用戶ID，而用戶ID來自認證信息。
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
# 1. 認證依賴：獲取當前用戶ID
def get_current_user_id():
# 假設從Token獲取用戶ID（實際場景用更安全的方式）
return "user_123" # 示例返回用戶ID
# 2. 數據庫查詢依賴：依賴用戶ID獲取用戶信息
async def get_user_by_id(user_id=Depends(get_current_user_id)):
db = Depends(get_async_db) # 依賴上面的數據庫連接
user = await db.query(f"SELECT * FROM users WHERE id={user_id}")
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
app = FastAPI()
@app.get("/user")
async def get_user(user=Depends(get_user_by_id)):
return user
這裏get_user_by_id同時依賴get_current_user_id（同步）和get_async_db（異步），FastAPI會自動按順序解析依賴項，確保依賴項的調用邏輯正確。
2. 異步任務中的依賴傳遞¶
當你需要在異步任務（比如後臺任務、消息隊列任務）中複用依賴項時，直接傳遞依賴項對象即可。FastAPI支持在任務函數中接收依賴項參數。
from fastapi import BackgroundTasks, Depends, FastAPI
# 1. 異步數據庫連接依賴
async def get_db():
db = await aio_db.connect()
return db
# 2. 後臺任務函數（異步）
async def send_welcome_email_task(db=Depends(get_db), user_id: str):
await db.execute(f"INSERT INTO emails (user_id, content) VALUES ({user_id}, 'Welcome!')")
app = FastAPI()
@app.post("/register")
async def register(
user_id: str,
background_tasks: BackgroundTasks,
db=Depends(get_db)
):
# 1. 先註冊用戶（假設db是同步操作）
await db.execute(f"INSERT INTO users (id) VALUES ({user_id})")
# 2. 添加異步後臺任務，傳遞db和user_id
background_tasks.add_task(
send_welcome_email_task, # 任務函數
user_id=user_id # 直接傳遞依賴項需要的參數
)
return {"message": "User registered, welcome email queued"}
這裏send_welcome_email_task接收db依賴項和user_id參數，FastAPI會在調用任務時自動注入db（已通過Depends(get_db)解析）。
3. 避免常見陷阱¶
- 忘記await異步依賴項：如果依賴項是
async def，但你直接return了協程對象（而非結果），會導致後續調用報錯。必須用Depends讓FastAPI自動await。 - 循環依賴：比如A依賴B，B又依賴A，FastAPI會拋出
RuntimeError。此時需重構代碼，拆分依賴項或使用緩存。 - 依賴項返回值類型：確保依賴項返回的是你需要的對象類型（比如數據庫連接、用戶信息），避免類型不匹配。
四、總結¶
FastAPI的異步依賴注入通過Depends()工具，讓我們能優雅地管理異步資源（如數據庫連接、認證信息），核心要點：
1. 依賴項函數：同步用普通函數，異步用async def，FastAPI自動處理await；
2. 嵌套依賴：通過Depends()聲明多層依賴，FastAPI按順序解析；
3. 任務傳遞：在異步任務中直接傳遞依賴項參數，確保資源複用。
掌握這些技巧後，你可以更高效地構建解耦、可擴展的FastAPI應用，尤其在處理異步任務時，能避免重複代碼和資源浪費。
小練習：嘗試給一個異步路由函數添加兩個依賴項（比如get_db和get_config），並在函數中使用它們完成一個簡單的數據庫查詢。