FastAPI依赖注入详解:Depends的基础与高级用法

什么是依赖注入?

想象一下,你正在写一个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 抽离出来?这会让你的代码更优雅、更强大!

小夜