FastAPI 内置了一套功能强大且简洁的依赖注入(Dependency Injection)系统。通过 Depends,你可以把公共逻辑(如数据库会话获取、鉴权、分页参数解析等)抽离成可复用的组件,FastAPI 会在请求到达路由函数前自动执行这些依赖,并把结果注入进来。本文系统讲解依赖注入的各类用法。

什么是依赖注入

依赖注入的核心思想是:路由函数只声明「我需要什么」,由框架负责「准备并传入」

1
2
3
4
5
6
7
8
9
10
from fastapi import FastAPI, Depends

app = FastAPI()

def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
return {"message": "items", "params": commons}

访问 /items/?q=python&skip=0&limit=10 时,FastAPI 会:

  1. 调用 common_parameters,把查询参数 qskiplimit 传进去;
  2. 把它的返回值赋给 commons
  3. 再调用 read_items,传入 commons

注意:common_parameters 的参数会被 FastAPI 自动识别为查询参数,因为它不在路径里。DependsQueryPath 不同——它本身不是查询参数,而是告诉框架「请先执行这个依赖」。


依赖可以被共享

多个路由复用同一个依赖,是依赖注入最常见的用途:

1
2
3
4
5
6
7
@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
return {"items": commons}

@app.get("/users/")
def read_users(commons: dict = Depends(common_parameters)):
return {"users": commons}

两处调用 common_parameters,逻辑只在依赖函数里维护一份,改一处即全局生效。

同一个请求内,对同一个依赖 FastAPI 默认会缓存结果。也就是说,若一个请求中多个地方依赖了同一个函数,它只会被执行一次,后续直接返回缓存值。


类作为依赖

除了函数,任何可调用对象(类、实例)都可以作为依赖。用类来定义依赖,结构更清晰:

1
2
3
4
5
6
7
8
9
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/")
def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

Depends(CommonQueryParams) 中传的参数是类本身(可调用),FastAPI 会实例化它,并把构造函数参数识别为查询参数。

简写形式

当依赖「类型」和「可调用对象」相同时,可以简写:

1
2
3
4
# 这三种写法等价
def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)): ...
def read_items(commons: CommonQueryParams = Depends()): ...
def read_items(commons: CommonQueryParams = Depends()): ...

Depends() 不传参时,FastAPI 会自动用参数的类型注解(这里是 CommonQueryParams)作为依赖。


子依赖(嵌套依赖)

依赖可以层层嵌套,FastAPI 会按顺序解析整个依赖树:

1
2
3
4
5
6
7
8
9
10
11
def query_extractor(q: str | None = None):
return q

def query_checker(q: str = Depends(query_extractor)):
if q == "admin":
raise ValueError("禁止使用保留关键字")
return q

@app.get("/search/")
def search(result: str = Depends(query_checker)):
return {"q": result}

执行顺序:query_extractorquery_checkersearch。如果某一层依赖抛出异常,请求会立即中断并返回对应错误。


全局依赖与路由组依赖

有些依赖(如鉴权、CORS 校验)需要应用到整个应用或一组路由上,无需在每个函数里写 Depends

应用级依赖

1
2
3
4
5
async def verify_token(x_token: str = Header(...)):
if x_token != "secret-token":
raise HTTPException(status_code=400, detail="X-Token 错误")

app = FastAPI(dependencies=[Depends(verify_token)])

这样所有路由都会先执行 verify_token,但它的返回值不会传入路由函数。

APIRouter 级依赖

1
2
3
4
5
6
7
8
9
from fastapi import APIRouter

admin_router = APIRouter(dependencies=[Depends(verify_token)])

@admin_router.get("/dashboard/")
def dashboard():
return {"data": "仅管理员可见"}

app.include_router(admin_router, prefix="/admin")

单个路由的专属依赖

1
2
3
@app.get("/items/", dependencies=[Depends(verify_token)])
def read_items():
return {"items": []}

dependencies 列表中的依赖会执行,但结果不注入函数——适合用于「拦截 / 校验」类需求。


使用 yield 的依赖(资源管理)

类似上下文管理器,依赖函数可以使用 yield 来在请求结束后执行清理逻辑:

1
2
3
4
5
6
7
8
9
10
11
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

@app.get("/users/")
def list_users(db = Depends(get_db)):
# 使用 db 查询
return db.execute("SELECT * FROM users").fetchall()

执行流程:

  1. get_db 执行到 yield db,把 db 注入路由函数;
  2. 路由函数执行完毕并返回响应;
  3. FastAPI 继续执行 yield 之后的 finally 代码,关闭连接。

这种方式非常适合管理数据库会话、Redis 连接、文件句柄等资源。也可以配合 with 语句使用:

1
2
3
def get_file():
with open("data.txt") as f:
yield f

实战示例:分页参数 + 鉴权

把依赖注入用到真实业务中。下面演示「分页依赖」和「当前用户依赖」的组合:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from fastapi import FastAPI, Depends, HTTPException, Header

app = FastAPI()

class Pagination:
def __init__(self, page: int = 1, size: int = 10):
self.page = page
self.size = size

def get_current_user(authorization: str = Header(...)):
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="无效的认证信息")
token = authorization[7:]
user = verify_jwt(token) # 假设的解密函数
if not user:
raise HTTPException(status_code=401, detail="用户不存在")
return user

@app.get("/posts/")
def list_posts(
pager: Pagination = Depends(Pagination),
current_user: dict = Depends(get_current_user),
):
return {
"page": pager.page,
"size": pager.size,
"user": current_user["username"],
}

一个路由函数同时依赖两个组件:Pagination 负责解析分页参数,get_current_user 负责鉴权。两者互不耦合,各自独立可测试、可复用。


小结

用法 关键写法 适用场景
函数依赖 Depends(func) 抽取公共参数 / 逻辑
类依赖 Depends(Class)Depends() 结构化参数对象
子依赖 依赖内部再调用 Depends 多层校验、拆分逻辑
全局依赖 FastAPI(dependencies=[...]) 全局拦截(鉴权、日志)
路由组依赖 APIRouter(dependencies=[...]) 一组接口共享前置逻辑
yield 依赖 函数中 yield 数据库会话等资源管理

依赖注入是 FastAPI 区别于其他框架的核心特性之一。善用 Depends,可以让接口代码保持「只关心业务本身」,把参数解析、权限校验、资源管理等工作交给可复用的依赖组件,显著提升代码的可维护性和可测试性。