Chapter 03 · 依赖注入:代码的乐高积木
📚 学完本章,你将能够:
- 解释"依赖注入"解决了什么实际问题(告别复制粘贴)
- 用
Depends把公共逻辑抽象成可复用函数 - 理解子依赖链:依赖的依赖的依赖
- 用
yield管理资源的创建和释放(如数据库连接) - 在测试中替换依赖,实现不启动数据库的单元测试
💡 场景引入
你的图书 API 有 10 个接口需要"当前登录用户"信息。你不得不在每个接口里重复写一遍"从请求头读取 Token → 解析 Token → 查数据库确认用户存在"的流程。
更麻烦的是,如果 Token 的解析逻辑变了(比如换了加密算法),你得改 10 个地方。依赖注入就是来解决这种重复问题的。
3.1 没有依赖注入的噩梦
@app.get("/books")
async def list_books(authorization: str = Header()):
token = authorization.replace("Bearer ", "")
user = verify_token(token) # 同样的代码
if not user:
raise HTTPException(401)
return books
@app.post("/books")
async def create_book(data: BookCreate, authorization: str = Header()):
token = authorization.replace("Bearer ", "")
user = verify_token(token) # 又写了一遍!
if not user:
raise HTTPException(401)
...
@app.delete("/books/{book_id}")
async def delete_book(book_id: int, authorization: str = Header()):
token = authorization.replace("Bearer ", "")
user = verify_token(token) # 第三遍!!
if not user:
raise HTTPException(401)
...
这段代码有三个问题:
- 重复:同样的逻辑写了 N 遍
- 难维护:改一处逻辑要改 N 个地方
- 难测试:每个接口都耦合了 Token 解析,测试时必须走完整流程
3.2 Depends:把公共逻辑抽出来
FastAPI 的 Depends 让你把公共逻辑写成一个函数,然后在需要的地方"注入":
from fastapi import Depends, HTTPException, Header
from typing import Annotated
async def get_current_user(authorization: Annotated[str, Header()]) -> User:
token = authorization.replace("Bearer ", "")
user = verify_token(token)
if not user:
raise HTTPException(401, detail="Invalid token")
return user
@app.get("/books")
async def list_books(user: Annotated[User, Depends(get_current_user)]):
# user 已经是验证通过的用户对象
return books
@app.post("/books")
async def create_book(data: BookCreate, user: Annotated[User, Depends(get_current_user)]):
...
@app.delete("/books/{book_id}")
async def delete_book(book_id: int, user: Annotated[User, Depends(get_current_user)]):
...
现在 Token 解析逻辑只写在一处。所有接口都通过 Depends(get_current_user) 复用这段逻辑。如果以后要改解析方式,只改 get_current_user 一个函数即可。
GET /booksget_current_userlist_books(user)3.3 子依赖链
依赖函数本身也可以依赖其他依赖函数。这叫做子依赖。
# 第一层:从 Header 提取 Token
async def get_token(auth: Annotated[str, Header(alias="Authorization")]) -> str:
if not auth.startswith("Bearer "):
raise HTTPException(401, "Must use Bearer token")
return auth[7:]
# 第二层:验证 Token 得到用户
async def get_current_user(token: Annotated[str, Depends(get_token)]) -> User:
user = verify_jwt(token)
if not user:
raise HTTPException(401, "Invalid or expired token")
return user
# 第三层:检查是否是管理员
async def require_admin(user: Annotated[User, Depends(get_current_user)]) -> User:
if user.role != "admin":
raise HTTPException(403, "Admin only")
return user
# 使用
@app.delete("/users/{user_id}")
async def ban_user(user: Annotated[User, Depends(require_admin)]):
...
当请求进入 /users/5 时,FastAPI 会自动按顺序执行:
get_token→ 提取字符串get_current_user→ 验证并得到 User 对象require_admin→ 检查角色- 最后进入
ban_user视图函数
同一个请求中,如果多个参数都 Depends(get_current_user),FastAPI 只会执行一次 get_current_user,然后缓存结果复用。这避免了重复查数据库。
3.4 yield:管理资源的创建和释放
数据库连接、文件句柄这类资源需要"用完后关闭"。yield 让依赖函数变成上下文管理器:
async def get_db():
db = create_database_connection() # 请求开始时创建连接
try:
yield db # 把连接交给视图函数使用
finally:
db.close() # 请求结束后关闭连接,即使出异常也会执行
yield 之前的代码在视图函数前执行,yield 之后的代码在视图函数后执行(包括异常时)。这非常适合数据库会话管理:
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
async_session = async_sessionmaker(engine, expire_on_commit=False)
async def get_db():
async with async_session() as session:
yield session # 视图函数使用 session
# 退出 async with 时自动关闭连接
3.5 依赖覆盖:测试的杀手锏
这是依赖注入最强大的特性。测试中你可以临时替换任何一个依赖,不用修改业务代码。
from fastapi.testclient import TestClient
# 测试中不用真的去验证 JWT
async def mock_current_user():
return User(id=1, username="tester", role="admin")
# 把真实的 get_current_user 换成假的
app.dependency_overrides[get_current_user] = mock_current_user
client = TestClient(app)
def test_delete_book():
response = client.delete("/books/1")
assert response.status_code == 204
# 测试结束后清理
app.dependency_overrides.clear()
这意味着:测试认证接口时,你不需要真的生成 Token;测试数据库操作时,你可以换成内存数据库——这些都会在第 9 章详细展开。
🎯 知识检测
Depends 的核心作用是什么?app.dependency_overrides[get_db] = fake_db 的作用是什么?.clear() 即可恢复,不会污染其他测试或生产代码。
本章小结
Depends 消灭重复代码
把"获取当前用户""打开数据库"这类公共逻辑写成函数,到处复用。
依赖可以无限嵌套
get_token → get_current_user → require_admin,像搭积木一样组合权限。
yield 管理资源生命周期
连接数据库 → yield 给路由使用 → 路由结束关闭连接。保证不泄漏。
依赖覆盖让测试更轻松
测试中替换真实依赖为 Mock,不用真的连数据库或发邮件。