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 会:
- 调用
common_parameters,把查询参数 q、skip、limit 传进去;
- 把它的返回值赋给
commons;
- 再调用
read_items,传入 commons。
注意:common_parameters 的参数会被 FastAPI 自动识别为查询参数,因为它不在路径里。Depends 和 Query、Path 不同——它本身不是查询参数,而是告诉框架「请先执行这个依赖」。
依赖可以被共享
多个路由复用同一个依赖,是依赖注入最常见的用途:
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_extractor → query_checker → search。如果某一层依赖抛出异常,请求会立即中断并返回对应错误。
全局依赖与路由组依赖
有些依赖(如鉴权、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)): return db.execute("SELECT * FROM users").fetchall()
|
执行流程:
get_db 执行到 yield db,把 db 注入路由函数;
- 路由函数执行完毕并返回响应;
- 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,可以让接口代码保持「只关心业务本身」,把参数解析、权限校验、资源管理等工作交给可复用的依赖组件,显著提升代码的可维护性和可测试性。