FastAPI依赖注入:简化代码结构的实用技巧

在FastAPI中,依赖注入(Dependency Injection,简称DI)是一个让代码更整洁、更灵活的核心特性。它帮助我们把代码中重复的逻辑(比如获取数据库连接、验证用户身份)集中管理,避免在每个接口里重复编写相同的代码,同时让测试和扩展变得更容易。

什么是依赖注入?

想象你写一个API接口,需要从数据库查询用户信息。如果每次都在接口里写“连接数据库→查询数据→关闭连接”,代码会很冗余。依赖注入的思路是:把“获取数据库连接”这个功能封装成一个独立的“依赖项”,然后在需要的接口里直接“请求”这个依赖项,而不用自己重复实现。

简单来说,依赖注入就是“把别人需要的东西(依赖)提前准备好,然后传给需要它的地方”。在FastAPI里,这个“准备”和“传递”的过程由FastAPI自动完成。

为什么FastAPI需要依赖注入?

  1. 代码复用:比如多个接口都需要数据库连接,用依赖注入可以只写一次连接逻辑,所有接口共享。
  2. 结构清晰:依赖项独立管理,接口函数只关心业务逻辑,不关心依赖如何获取。
  3. 测试方便:测试时可以用“模拟的依赖”代替真实依赖(比如用假数据代替真实数据库),让单元测试更简单。
  4. 资源管理:比如数据库连接、认证信息等资源的创建和释放,由依赖注入统一处理,避免内存泄漏。

核心用法:定义和使用依赖项

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
- 依赖项的执行顺序由依赖链决定,确保所有前置依赖都已准备好。

依赖注入的优势

  1. 减少重复代码:相同逻辑(如数据库连接)只需写一次,所有接口复用。
  2. 便于测试:单元测试时可替换依赖项为模拟对象(如用unittest.mock模拟数据库连接)。
  3. 自动资源管理:通过yield处理资源(如连接关闭),避免手动管理。
  4. 与FastAPI文档集成:依赖项的参数和返回值会自动显示在Swagger UI中,方便接口调试。

最佳实践

  • 单一职责:每个依赖项只做一件事(如只负责数据库连接或只负责用户认证)。
  • 避免过度依赖:复杂依赖链可能降低可读性,可拆分为多个小依赖项。
  • 异步依赖:如果依赖项是异步操作(如异步数据库查询),用async def定义依赖项,FastAPI会自动处理。

依赖注入让FastAPI代码更模块化、更易维护。掌握它后,你会发现处理重复逻辑和复杂业务变得异常轻松。接下来可以尝试在自己的项目中,把数据库连接、认证逻辑等抽象为依赖项,体验代码简化的快感!

小夜