07 依赖注入(Depends)
本章目标:理解 FastAPI 的依赖注入机制。这是全教程最重要的一章——后面连接数据库、实现登录鉴权,全靠它。
1. 从一个重复代码的问题说起
假设很多接口都需要分页参数:
@app.get("/articles")
def list_articles(page: int = 1, size: int = 10): ...
@app.get("/users")
def list_users(page: int = 1, size: int = 10): ...
@app.get("/comments")
def list_comments(page: int = 1, size: int = 10): ...
page、size 写了三遍,将来要加校验(size 最大 100)就得改三处。依赖注入可以把这段公共逻辑抽出来。
2. 基本用法:Depends
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
# 1. 把公共逻辑写成一个普通函数(叫「依赖函数」)
def pagination(page: int = 1, size: int = 10):
if size > 100:
size = 100
return {"page": page, "size": size}
# 2. 在接口中用 Depends 声明「我需要它」
@app.get("/articles")
def list_articles(pg: Annotated[dict, Depends(pagination)]):
return {"resource": "articles", **pg}
@app.get("/users")
def list_users(pg: Annotated[dict, Depends(pagination)]):
return {"resource": "users", **pg}
执行流程:请求到达 /articles?page=2&size=10 时,FastAPI 会:
- 看到该接口依赖
pagination - 先调用
pagination——它的参数(page、size)同样按查询参数规则从请求中解析、校验 - 把
pagination的返回值作为pg传给接口函数
所以在 /docs 里,/articles 接口仍然显示 page、size 两个查询参数——依赖函数的参数会「透传」到接口文档中。
心智模型:
Depends(f)= 「调用我之前,先帮我运行 f,把结果给我」。
简化重复的类型注解
同一个依赖到处用时,给它起个类型别名:
PageParams = Annotated[dict, Depends(pagination)]
@app.get("/articles")
def list_articles(pg: PageParams): ...
@app.get("/users")
def list_users(pg: PageParams): ...
这是官方推荐的写法,后面我们会定义 SessionDep(数据库会话)和 CurrentUser(当前登录用户)两个常用别名。
3. 依赖可以嵌套
依赖函数自己也可以有依赖,FastAPI 会自动解析整棵依赖树:
def get_token(authorization: str | None = None):
# 从请求头/参数里拿 token(示意)
return authorization
def get_current_user(token: Annotated[str | None, Depends(get_token)]):
if token is None:
raise HTTPException(status_code=401, detail="请先登录")
return {"id": 1, "username": "xiaoming"} # 假装根据 token 查到了用户
@app.get("/me")
def read_me(user: Annotated[dict, Depends(get_current_user)]):
return user
注意 get_current_user 里直接 raise HTTPException——依赖里抛异常,接口函数根本不会执行。这正是鉴权的实现方式:把「验证登录」做成依赖,需要登录的接口挂上它即可。
4. yield 依赖:带清理动作的依赖(重点!)
有些资源用完必须释放,比如数据库连接。依赖函数里用 yield 可以实现「用前创建、用后清理」:
def get_db():
db = create_connection() # 伪代码:创建数据库连接
try:
yield db # 把 db 交给接口函数使用
finally:
db.close() # 接口执行完毕后(无论成功还是报错),关闭连接
执行顺序:
请求到达
→ 执行 yield 之前的代码(创建连接)
→ 接口函数拿到 db,执行业务逻辑
→ 响应返回
→ 执行 finally 里的代码(关闭连接)
这就是第四部分连接 SQLAlchemy 的核心模式,先在这里混个眼熟:
# 剧透:第四部分的 get_db 长这样
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
5. 类作为依赖(了解)
依赖不一定是函数,类也行——FastAPI 会用请求参数实例化它:
class Pagination:
def __init__(self, page: int = 1, size: int = 10):
self.page = page
self.size = min(size, 100)
self.offset = (page - 1) * self.size # 顺便算好数据库偏移量
@app.get("/articles")
def list_articles(pg: Annotated[Pagination, Depends(Pagination)]):
return {"page": pg.page, "size": pg.size, "offset": pg.offset}
类依赖的好处是能带方法和计算属性,适合参数较多的场景。
6. 不需要返回值的依赖:路由级/全局挂载
有时依赖只做检查,不需要返回值(比如验证 API Key)。可以挂在装饰器上:
def verify_api_key(x_api_key: Annotated[str | None, Header()] = None):
if x_api_key != "secret123":
raise HTTPException(status_code=403, detail="无效的 API Key")
# 挂在单个接口上
@app.get("/admin/stats", dependencies=[Depends(verify_api_key)])
def admin_stats():
return {"users": 100}
# 或挂在整个应用上(所有接口都要过这一关)
app = FastAPI(dependencies=[Depends(verify_api_key)])
(Header() 用于读取请求头,用法和 Query 一致,从 fastapi 导入。)
7. 小结:依赖注入解决了什么
| 场景 | 用依赖注入的方式 |
|---|---|
| 公共参数(分页、排序) | 抽成依赖函数,处处复用 |
| 数据库会话 | yield 依赖,自动开关连接 |
| 登录鉴权 | 依赖里验证 token,失败直接 401 |
| 权限检查 | 依赖里检查角色,失败 403 |
核心记忆点:
Depends(f):执行接口前先执行 f,返回值注入进来- 依赖可嵌套,形成依赖树
- 依赖里
raise HTTPException可拦截请求 yield依赖 = 请求前准备资源 + 请求后清理资源- 用
Annotated类型别名消除重复
FastAPI 基础到此告一段落!接下来进入数据库的世界:03-SQLAlchemy基础/01-数据库与ORM入门