FastAPI 依赖注入

在 FastAPI 中,依赖注入(Dependency Injection, DI) 是其最强大、最具标志性的特性。它允许你声明你的代码所需的“依赖项”(如数据库连接、当前用户、分页参数等),FastAPI 会在运行时自动为你解析、执行并注入这些依赖。

使用依赖注入的核心好处是:代码高度复用、逻辑解耦、以及极其方便的单元测试

下面我将从基础语法核心应用场景进阶技巧,为你系统梳理 FastAPI 的依赖注入系统。


一、 基础语法:Depends

依赖项本质上就是一个普通的 Python 函数(或类)。你只需在路由函数的参数中使用 Depends() 包裹它即可。

纯文本
from fastapi import FastAPI, Depends

app = FastAPI()

# 1. 定义一个依赖项 (普通的 Python 函数)
def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 2. 在路由中注入依赖
@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
    # commons 就是 common_parameters 函数的返回值
    return {"message": "获取物品列表", "params": commons}

@app.get("/users/")
def read_users(commons: dict = Depends(common_parameters)):
    # 复用了同一个依赖项!
    return {"message": "获取用户列表", "params": commons}

💡 提示:FastAPI 会自动读取 common_parameters 的参数类型提示,并在 Swagger UI 中将其显示为查询参数。


二、 四大核心应用场景 🌟

1. 数据库会话管理 (使用 yield 的生成器依赖)

这是 FastAPI 中最经典、最重要的依赖注入模式。使用 yield 可以确保在请求处理完毕后,自动执行清理逻辑(如关闭数据库连接或回滚事务)。

纯文本
from fastapi import Depends, FastAPI
# 假设使用 SQLAlchemy
# from sqlalchemy import create_engine
# from sqlalchemy.orm import sessionmaker, Session

# engine = create_engine("sqlite:///./test.db")
# SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 1. 定义依赖项 (生成器函数)
def get_db():
    # db = SessionLocal()  # 创建数据库会话
    db = "MockDBSession"   # 这里用字符串模拟
    try:
        yield db           # 将 db 注入给路由函数
    finally:
        # 请求结束后,无论是否发生异常,都会执行这里
        # db.close()       # 确保连接被安全关闭
        print("数据库连接已关闭")

# 2. 在路由中使用
@app.get("/users/")
def read_users(db: str = Depends(get_db)):
    # users = db.query(User).all()
    return {"db_session": db, "users": ["Alice", "Bob"]}

2. 身份验证与权限校验 (Security)

依赖项非常适合用来拦截请求,校验 Token 或权限。如果校验失败,直接抛出 HTTPException,路由函数根本不会被执行。

纯文本
from fastapi import Depends, HTTPException, status, Header

# 模拟的验证逻辑
def get_current_user(x_token: str = Header(...)):
    if x_token != "my-super-secret-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return {"username": "admin", "role": "superuser"}

# 只有携带正确 Token 的请求才能进入此路由
@app.get("/protected-data")
def read_protected_data(current_user: dict = Depends(get_current_user)):
    return {"message": f"欢迎, {current_user['username']}!", "data": "机密信息"}

3. 依赖的依赖 (嵌套依赖)

依赖项本身也可以声明它们自己的依赖项。FastAPI 会自动构建依赖树,并确保每个依赖只执行一次(即使被多次引用)。

纯文本
def get_query(q: str | None = None):
    return q

def get_token_header(x_token: str = Header(...)):
    return x_token

# 这个依赖项依赖于上面两个依赖项
def get_current_user_with_query(
    q: str = Depends(get_query),
    token: str = Depends(get_token_header)
):
    return {"query": q, "token": token}

@app.get("/advanced/")
def advanced_route(user_info: dict = Depends(get_current_user_with_query)):
    return user_info

4. 路由级或全局依赖 (Dependencies at the router/app level)

如果你希望某个依赖项应用于整个路由器整个应用,但不需要将其作为参数传递给每个函数,可以在 APIRouterFastAPI 实例中声明。

纯文本
from fastapi import APIRouter, Depends

# 定义一个记录日志的依赖
def log_request():
    print("📝 收到请求")

# 将此依赖应用到该路由器的所有路由上
router = APIRouter(dependencies=[Depends(log_request)])

@router.get("/items/")
def get_items():
    return ["item1", "item2"] # 访问此路由时,会自动先执行 log_request

app.include_router(router, prefix="/api")

三、 进阶特性:依赖覆盖 (Override Dependencies)

这是依赖注入系统最强大的杀手锏,专门用于单元测试。在测试时,你可以用 Mock(模拟)函数替换真实的依赖项(如真实的数据库连接或外部 API 调用),而无需修改任何业务代码。

纯文本
# main.py
def get_db():
    return "RealProductionDatabase"

@app.get("/items/")
def read_items(db: str = Depends(get_db)):
    return {"db": db}
纯文本
# test_main.py
from fastapi.testclient import TestClient
from main import app, get_db

# 1. 定义一个模拟的依赖项
def override_get_db():
    return "MockTestDatabase"

# 2. 在测试前覆盖依赖
app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_read_items():
    response = client.get("/items/")
    assert response.status_code == 200
    assert response.json() == {"db": "MockTestDatabase"} # 成功使用了 Mock 数据

# 3. 测试结束后清除覆盖 (可选,但推荐)
app.dependency_overrides.clear()

四、 类作为依赖项 (Classes as Dependencies)

除了函数,你也可以使用作为依赖项。FastAPI 会自动实例化该类,并将其作为依赖项注入。这在封装复杂逻辑(如分页配置、复杂的权限校验器)时非常有用。

纯文本
from fastapi import Depends, Query

class Paginator:
    def __init__(self, skip: int = Query(0, ge=0), limit: int = Query(10, le=100)):
        self.skip = skip
        self.limit = limit

@app.get("/users/")
def get_users(pagination: Paginator = Depends(Paginator)):
    # FastAPI 会自动调用 Paginator(skip=..., limit=...)
    return {"skip": pagination.skip, "limit": pagination.limit}

五、 最佳实践与避坑指南

  1. 始终使用类型提示 (Type Hints):FastAPI 依赖类型提示来解析参数和生成文档。没有类型提示的依赖项将无法在 Swagger UI 中正确显示。
  2. 生成器依赖 (yield) 必须配合 try...finally:确保无论业务逻辑是否抛出异常,资源(如 DB 连接、文件句柄)都能被正确清理。
  3. 避免在依赖项中产生副作用:依赖项应该尽量是“纯”的,或者只负责获取资源/校验权限。复杂的业务逻辑应保留在路由函数或 Service 层中。
  4. 不要滥用全局依赖:虽然 app.include_router(..., dependencies=[...]) 很方便,但过度使用会导致代码隐式行为过多,降低可读性。优先使用显式的函数参数注入。

总结:依赖注入速查表

场景推荐模式
提取公共查询参数普通函数 + Depends(func)
数据库连接/资源清理生成器函数 (yield) + try...finally
权限校验/Token 验证抛出 HTTPException 的函数 + Depends
封装复杂配置 (如分页)使用 Class + Depends(ClassName)
编写单元测试 (Mock 数据)app.dependency_overrides[real_dep] = mock_dep

依赖注入是 FastAPI 从“好用”迈向“企业级工程化”的关键。你可以尝试将你项目中的数据库连接或用户认证逻辑重构为 Depends 模式,体验代码解耦带来的清爽感!

如果你在实现具体的依赖(例如 JWT Token 解析多数据库动态切换)时遇到困难,随时告诉我,我可以提供完整的代码模板!