FastAPI 的依赖注入(Dependency Injection, DI) 是其最强大、最核心的特性之一。它允许你声明代码运行所需的“依赖”(如数据库连接、当前用户、配置信息等),FastAPI 会自动为你解析、执行并注入这些依赖。
这种设计不仅让代码更简洁、可复用,还极大地简化了单元测试。
以下是 FastAPI 依赖注入的核心概念和常用用法:
1. 基础用法:使用 Depends
依赖通常是一个普通的 Python 函数(或类)。你可以使用 Depends 将其注入到路径操作函数中。
from fastapi import FastAPI, Depends
app = FastAPI()
# 1. 定义一个依赖函数
def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
# 2. 在路径操作中使用 Depends 注入
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
return commonsFastAPI 会自动调用 common_parameters,解析其中的查询参数,并将返回值传递给 commons。
2. 现代推荐写法:使用 Annotated (FastAPI 0.95.0+)
官方目前强烈推荐使用 typing.Annotated,它可以让类型提示和依赖声明分离,代码更清晰,且兼容更多静态类型检查工具。
from typing import Annotated
from fastapi import FastAPI, Depends
app = FastAPI()
def get_current_user(token: str):
# 模拟验证 token 并返回用户
return {"username": "admin", "token": token}
# 使用 Annotated 声明依赖
@app.get("/users/me")
async def read_users_me(
current_user: Annotated[dict, Depends(get_current_user)]
):
return current_user3. 类作为依赖
依赖不仅可以是函数,也可以是类。FastAPI 会实例化该类,并将其作为依赖项注入。
from typing import Annotated
from fastapi import FastAPI, Depends
app = FastAPI()
class CommonQueryParams:
def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends()]):
# 注意:这里 Depends() 里面可以为空,FastAPI 会根据类型提示自动推断
return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}4. 带有 yield 的依赖(处理清理工作)
这是 FastAPI 依赖注入最实用的场景之一。如果你的依赖需要在请求结束后执行清理操作(如关闭数据库连接、释放文件句柄),可以使用 yield 代替 return。
from typing import Annotated
from fastapi import FastAPI, Depends
app = FastAPI()
# 模拟数据库会话
class FakeDBSession:
def __init__(self):
print("🔗 数据库连接已建立")
def close(self):
print("🔌 数据库连接已关闭")
def get_db():
db = FakeDBSession()
try:
yield db # 将 db 注入给路径操作函数
finally:
db.close() # 请求结束后,无论是否发生异常,都会执行这里
@app.get("/items/")
async def read_items(db: Annotated[FakeDBSession, Depends(get_db)]):
return {"status": "使用数据库查询数据"}运行结果:先打印“建立连接”,处理请求,最后打印“关闭连接”。
5. 依赖缓存 (Dependency Caching)
在同一个请求中,如果你多次声明了同一个依赖,FastAPI 默认只会执行一次,并缓存其结果。这可以显著节省资源(例如避免多次查询数据库)。
def get_query(q: str | None = None):
print("执行了 get_query") # 这个函数在一次请求中只会打印一次
return {"q": q}
@app.get("/items/")
async def read_items(
q1: Annotated[dict, Depends(get_query)],
q2: Annotated[dict, Depends(get_query)] # 不会再次执行 get_query
):
return {"q1": q1, "q2": q2}如果你希望每次都重新执行,可以设置 Depends(get_query, use_cache=False)。
6. 子依赖 (Sub-dependencies)
依赖可以拥有自己的依赖。FastAPI 会自动构建一个依赖图(Dependency Graph),并按正确的顺序解析它们。
def verify_token(x_token: str):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
return x_token
def verify_key(x_key: str):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
# get_header 依赖于 verify_token 和 verify_key
def get_header(
x_token: Annotated[str, Depends(verify_token)],
x_key: Annotated[str, Depends(verify_key)]
):
return {"x_token": x_token, "x_key": x_key}
@app.get("/items/")
async def read_items(header: Annotated[dict, Depends(get_header)]):
return header7. 覆盖依赖 (Dependency Overrides) – 测试神器
在编写单元测试时,你通常不想连接真实的数据库或调用真实的外部 API。FastAPI 提供了 app.dependency_overrides,允许你用 Mock 函数替换真实的依赖。
from fastapi.testclient import TestClient
# 真实的依赖
def get_db():
return "Real Database Connection"
@app.get("/data")
async def get_data(db: Annotated[str, Depends(get_db)]):
return {"db": db}
# --- 测试代码 ---
client = TestClient(app)
def override_get_db():
return "Mock Database Connection"
# 覆盖依赖
app.dependency_overrides[get_db] = override_get_db
def test_read_data():
response = client.get("/data")
assert response.json() == {"db": "Mock Database Connection"}
# 测试结束后,建议清除覆盖
app.dependency_overrides.clear()常见应用场景总结
- 数据库会话管理:使用
yield获取和关闭 SQLAlchemy/SQLModel 会话。 - 身份验证与授权:解析 JWT Token,获取当前登录用户 (
get_current_user)。 - 速率限制 (Rate Limiting):检查 IP 或用户的请求频率。
- 共享业务逻辑:将复杂的参数校验或数据预处理提取为独立的依赖函数。
如果你有具体的场景(比如:如何结合 SQLAlchemy 写依赖,或者如何做 JWT 认证),可以告诉我,我可以提供更针对性的代码示例!