在FastAPI中,依赖注入(Dependency Injection,简称DI)是一个让代码更整洁、更灵活的核心特性。它帮助我们把代码中重复的逻辑(比如获取数据库连接、验证用户身份)集中管理,避免在每个接口里重复编写相同的代码,同时让测试和扩展变得更容易。
什么是依赖注入?¶
想象你写一个API接口,需要从数据库查询用户信息。如果每次都在接口里写“连接数据库→查询数据→关闭连接”,代码会很冗余。依赖注入的思路是:把“获取数据库连接”这个功能封装成一个独立的“依赖项”,然后在需要的接口里直接“请求”这个依赖项,而不用自己重复实现。
简单来说,依赖注入就是“把别人需要的东西(依赖)提前准备好,然后传给需要它的地方”。在FastAPI里,这个“准备”和“传递”的过程由FastAPI自动完成。
为什么FastAPI需要依赖注入?¶
- 代码复用:比如多个接口都需要数据库连接,用依赖注入可以只写一次连接逻辑,所有接口共享。
- 结构清晰:依赖项独立管理,接口函数只关心业务逻辑,不关心依赖如何获取。
- 测试方便:测试时可以用“模拟的依赖”代替真实依赖(比如用假数据代替真实数据库),让单元测试更简单。
- 资源管理:比如数据库连接、认证信息等资源的创建和释放,由依赖注入统一处理,避免内存泄漏。
核心用法:定义和使用依赖项¶
FastAPI通过Depends类实现依赖注入。我们需要先定义一个“依赖项函数”,然后在接口函数中用Depends(依赖项)声明依赖。
1. 基础依赖项:获取数据库连接¶
假设我们需要在多个接口中使用数据库连接,传统写法会重复创建连接。用依赖注入后,我们可以这样做:
from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, Session
# 1. 定义数据库连接依赖项
def get_db():
# 创建数据库连接(每次请求时执行)
engine = create_engine("sqlite:///test.db")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db = SessionLocal()
try:
yield db # 返回连接,供接口使用
finally:
db.close() # 请求结束后关闭连接
# 2. 定义数据库模型
Base = declarative_base()
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
# 3. 创建FastAPI应用
app = FastAPI()
# 4. 在接口中使用依赖项
@app.get("/items/{item_id}")
def read_item(item_id: int, db: Session = Depends(get_db)):
# 直接使用db(数据库连接)查询数据
item = db.query(Item).filter(Item.id == item_id).first()
return {"id": item.id, "name": item.name}
关键点:
- get_db是依赖项函数,用yield返回连接,确保请求结束后自动关闭连接(类似上下文管理器)。
- 接口函数read_item中,db: Session = Depends(get_db)声明依赖,FastAPI会自动调用get_db()获取连接并传入。
2. 依赖项带参数:根据用户ID获取用户信息¶
如果依赖项需要参数(比如根据用户ID查询用户),可以在依赖项函数中定义参数,FastAPI会自动解析路由参数或查询参数。
from fastapi import Depends, HTTPException
from pydantic import BaseModel
# 模拟用户数据(实际项目中可能来自数据库)
fake_users_db = {
1: {"id": 1, "name": "Alice"},
2: {"id": 2, "name": "Bob"}
}
# 1. 定义依赖项:获取用户信息
def get_user(user_id: int):
user = fake_users_db.get(user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
# 2. 在接口中使用依赖项(依赖项需要参数)
@app.get("/users/{user_id}")
def read_user(user_id: int, user: dict = Depends(get_user)):
return user
关键点:
- get_user依赖项需要user_id参数,FastAPI会自动从路由路径参数{user_id}中获取user_id的值,传给get_user。
- 如果用户不存在,直接抛出HTTP异常,FastAPI会自动返回错误响应。
3. 嵌套依赖:依赖项依赖其他依赖项¶
如果一个依赖项需要另一个依赖项,可以形成“嵌套依赖”,FastAPI会按顺序解析。
# 1. 基础依赖项:获取数据库连接
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# 2. 第二个依赖项:获取用户信息(依赖数据库连接)
def get_current_user(db: Session = Depends(get_db)):
# 假设用户信息存在数据库中,这里从db查询用户ID为1的用户
user = db.query(User).filter(User.id == 1).first()
return user
# 3. 第三个依赖项:验证用户权限(依赖用户信息)
def get_admin(user: dict = Depends(get_current_user)):
if user.get("role") != "admin":
raise HTTPException(status_code=403, detail="Not admin")
return user
# 4. 接口使用嵌套依赖
@app.get("/admin")
def admin_page(admin: dict = Depends(get_admin)):
return {"message": "Admin page accessed", "user": admin}
关键点:
- get_admin依赖get_current_user,而get_current_user依赖get_db,FastAPI会先解析get_db,再解析get_current_user,最后解析get_admin。
- 依赖项的执行顺序由依赖链决定,确保所有前置依赖都已准备好。
依赖注入的优势¶
- 减少重复代码:相同逻辑(如数据库连接)只需写一次,所有接口复用。
- 便于测试:单元测试时可替换依赖项为模拟对象(如用
unittest.mock模拟数据库连接)。 - 自动资源管理:通过
yield处理资源(如连接关闭),避免手动管理。 - 与FastAPI文档集成:依赖项的参数和返回值会自动显示在Swagger UI中,方便接口调试。
最佳实践¶
- 单一职责:每个依赖项只做一件事(如只负责数据库连接或只负责用户认证)。
- 避免过度依赖:复杂依赖链可能降低可读性,可拆分为多个小依赖项。
- 异步依赖:如果依赖项是异步操作(如异步数据库查询),用
async def定义依赖项,FastAPI会自动处理。
依赖注入让FastAPI代码更模块化、更易维护。掌握它后,你会发现处理重复逻辑和复杂业务变得异常轻松。接下来可以尝试在自己的项目中,把数据库连接、认证逻辑等抽象为依赖项,体验代码简化的快感!