在 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_info4. 路由级或全局依赖 (Dependencies at the router/app level)
如果你希望某个依赖项应用于整个路由器或整个应用,但不需要将其作为参数传递给每个函数,可以在 APIRouter 或 FastAPI 实例中声明。
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}五、 最佳实践与避坑指南
- 始终使用类型提示 (Type Hints):FastAPI 依赖类型提示来解析参数和生成文档。没有类型提示的依赖项将无法在 Swagger UI 中正确显示。
- 生成器依赖 (
yield) 必须配合try...finally:确保无论业务逻辑是否抛出异常,资源(如 DB 连接、文件句柄)都能被正确清理。 - 避免在依赖项中产生副作用:依赖项应该尽量是“纯”的,或者只负责获取资源/校验权限。复杂的业务逻辑应保留在路由函数或 Service 层中。
- 不要滥用全局依赖:虽然
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 解析 或 多数据库动态切换)时遇到困难,随时告诉我,我可以提供完整的代码模板!