在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），并在函数中使用它们完成一个简单的数据库查询。