跳到主要内容

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): ...

pagesize 写了三遍,将来要加校验(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 会:

  1. 看到该接口依赖 pagination
  2. 先调用 pagination——它的参数(page、size)同样按查询参数规则从请求中解析、校验
  3. 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入门