什么是依赖注入?¶
想象一下,你正在写一个API,需要获取用户信息。这个用户信息可能来自数据库、缓存或者认证服务。如果每个API端点都自己写获取用户信息的代码,不仅重复,还会导致修改时需要改多处(比如换数据库)。
依赖注入(Dependency Injection,简称DI)的核心思想是:把“依赖”(比如数据库连接、用户信息)通过系统自动“注入”到函数中,而不是让函数自己去创建或获取依赖。这样代码更易复用、解耦,维护起来更方便。
在FastAPI中,实现依赖注入的核心工具是 Depends——它就像一个“中介”,帮你把需要的依赖对象“送”到函数面前。
FastAPI中的Depends基础¶
使用 Depends 分两步:定义依赖函数和在路径函数中声明依赖。
1. 定义依赖函数¶
依赖函数负责“生产”你需要的对象(比如数据库连接、用户信息)。例如,模拟一个数据库连接:
from fastapi import FastAPI, Depends
# 定义依赖函数:返回数据库连接对象
def get_db():
# 这里返回模拟的数据库连接(实际项目中可能是真实连接)
return {"connection": "MySQL"} # 假设是数据库连接字典
2. 在路径函数中使用Depends¶
用 Depends(依赖函数) 声明路径函数需要的依赖,FastAPI会自动调用依赖函数并注入结果:
app = FastAPI()
@app.get("/users/me")
def read_current_user(db: dict = Depends(get_db)):
# db就是依赖函数get_db返回的数据库连接对象
return {"db_connection": db, "message": "当前用户信息"}
当访问 /users/me 时,FastAPI会先调用 get_db() 获取数据库连接,再把它作为 db 参数传给 read_current_user。
依赖注入的参数传递¶
依赖函数也可以接收参数,这些参数可以是路径参数、查询参数等,由FastAPI自动解析。
示例:根据用户ID获取用户信息¶
假设需要根据路径参数 user_id 查询用户,用户信息的获取依赖于数据库连接:
from fastapi import Depends, HTTPException
# 模拟用户数据
mock_users = {1: {"id": 1, "name": "Alice"}, 2: {"id": 2, "name": "Bob"}}
# 依赖函数:根据user_id查询用户(user_id来自路径参数)
def get_user(user_id: int = Depends()):
if user_id not in mock_users:
raise HTTPException(status_code=404, detail="用户不存在")
return mock_users[user_id]
# 路径函数:声明依赖
@app.get("/users/{user_id}")
def read_user(user: dict = Depends(get_user)):
return {"user": user}
这里:
- get_user 函数通过 Depends() 捕获路径参数 user_id(因为 user_id 是路径参数,FastAPI会自动解析)。
- user_id 作为参数传入 get_user,返回对应用户信息。
高级用法¶
依赖注入不仅能解决基础复用问题,还支持嵌套依赖、缓存、异步等场景。
1. 嵌套依赖(依赖的依赖)¶
一个依赖可以依赖另一个依赖,FastAPI会自动按顺序解析。例如,先获取数据库连接,再根据连接获取用户:
def get_db():
return {"connection": "MySQL"} # 第一步:获取数据库连接
def get_user_by_db(db: dict = Depends(get_db), user_id: int = 1):
# 第二步:用数据库连接查询用户(这里user_id简化为固定值,实际可从参数获取)
if user_id not in mock_users:
raise HTTPException(status_code=404, detail="用户不存在")
return mock_users[user_id]
@app.get("/users/{user_id}")
def read_user(user: dict = Depends(get_user_by_db)):
return {"user": user}
执行顺序:get_user_by_db 依赖 get_db,所以FastAPI先调用 get_db 获取数据库连接,再传入 get_user_by_db 查询用户。
2. 依赖缓存(单例模式)¶
如果依赖函数需要复用(比如数据库连接池),可以用 lru_cache 实现“单例”,避免重复初始化:
from functools import lru_cache
@lru_cache() # 缓存依赖函数结果
def get_db():
print("初始化数据库连接...") # 仅打印一次,后续复用结果
return {"connection": "MySQL"}
@app.get("/users/me")
def read_user(db: dict = Depends(get_db)):
return {"db": db}
多次调用 /users/me 时,get_db 只会打印一次“初始化数据库连接…”,后续直接返回缓存的连接对象。
3. 异步依赖¶
如果依赖函数是异步的(比如调用异步数据库),Depends 也支持:
import asyncio
async def get_async_db():
await asyncio.sleep(0.1) # 模拟异步数据库操作
return {"connection": "AsyncMySQL"}
@app.get("/async/users")
async def async_read_user(db: dict = Depends(get_async_db)):
return {"async_db": db}
注意:异步路径函数(async def)才能处理异步依赖的返回值。
4. 依赖参数验证¶
结合Pydantic模型,依赖函数可以自动验证参数合法性:
from pydantic import BaseModel
class UserFilter(BaseModel):
user_id: int = 1 # 定义参数类型和默认值
def get_user_filter(filter: UserFilter = Depends()):
return filter # 返回验证后的参数
@app.get("/validated/users")
def validated_user(filter: UserFilter = Depends(get_user_filter)):
return {"user_id": filter.user_id}
如果请求参数不符合 UserFilter 定义(比如 user_id 不是整数),FastAPI会自动返回 422 Validation Error。
依赖注入的核心优势¶
- 代码复用:同一个依赖被多个路径函数共享,避免重复代码。
- 解耦:路径函数只关心业务逻辑,依赖的获取交给系统。
- 易测试:测试时可用模拟对象(mock)替换真实依赖(如测试用的假数据库)。
- 可扩展:新增依赖时,只需修改依赖函数,无需改动所有调用它的路径函数。
总结¶
Depends 是FastAPI中管理依赖关系的核心工具,从基础的依赖定义、参数传递,到高级的嵌套依赖、异步依赖,掌握这些用法能让你的API代码更清晰、更易维护。通过合理设计依赖,你可以把复杂的逻辑拆解成独立的模块,让项目结构更健壮。
下次开发FastAPI应用时,不妨思考:哪些重复的依赖逻辑可以通过 Depends 抽离出来?这会让你的代码更优雅、更强大!