介绍
依赖就是 可复用的函数/类，把 公共逻辑抽出来，某个路由、接口需要调用时，作为 函数参数传入
注意：只能应用在 路由函数的形参
类比
对于Java程序员，类似Spring的AOP/拦截器
应用场景
- 权限校验
- 参数预处理，如：统一处理分页参数
page当前页数、size每页记录数 - 数据库会话
- 公共请求头读取
- 限流
1. 基础语法
步骤：
- 定义依赖函数
- 在需要的路由函数上指定参数，并给参数赋默认值：
Depends(依赖函数名)
例子
from fastapi import FastAPI, Depends
app = FastAPI()
# 定义依赖函数
def test_depend():
return "测试依赖"
# 在参数中指定默认值 Depends()，fastapi会自动注入
@app.get('/test')
async def get_student(content: str = Depends(test_depend)):
return {
"code": 200,
"msg": content
}
测试
浏览器发请求：http://127.0.0.1:8000/test ，显示如下：
{
"code": 200,
"msg": 测试依赖
}
先执行依赖函数 test_depend() ，然后 依赖函数 的 返回值 赋值给 形参
2. 依赖可以接收请求参数
依赖可以直接拿到 query请求参数、path路径参数
例子
from fastapi import FastAPI, Depends
app = FastAPI()
# 处理分页参数的依赖函数
def parse_page_param(page:int = 1 , size:int = 10):
'''
解析分页参数
:param page: 当前页数
:param size: 每页的记录数
:return:
'''
return {
"page":page,
"size":size
}
# 在参数中指定默认值 Depends()
@app.get('/students')
async def get_student(page: str = Depends(parse_page_param)):
print("page:", page)
return {
"code": 200,
"msg": "学习依赖"
}
测试
浏览器发请求：http://127.0.0.1:8000/students?page=1&size=10 ，控制台打印如下：
page: {'page': 1, 'size': 10}
3. 案例：鉴权
from fastapi import FastAPI, Depends, HTTPException
from starlette import status
app = FastAPI()
# 处理分页参数的依赖类
def check_token(token:str):
"""
鉴权函数，如果当前登录用户（token）是admin，就认为鉴权通过
:param token:
:return:
"""
if token == "admin":
return token
else:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="未授权")
# 在参数中指定默认值 Depends()
@app.get('/students')
async def get_student(token: str = Depends(check_token)):
return {
"code": 200,
"msg": f"鉴权通过，当前用户是：{token}"
}
测试：鉴权通过
浏览器访问：http://127.0.0.1:8000/students?token=admin ，浏览器显示如下：
{
"code": 200,
"msg": "鉴权通过，当前用户是：admin"
}
测试：鉴权失败
浏览器访问：http://127.0.0.1:8000/students?token=lili ，浏览器显示如下：
{
"detail": "未授权"
}
高级用法
类作为依赖
类也可以做依赖，FastAPI会自动实例化：
from fastapi import FastAPI, Depends
app = FastAPI()
# 处理分页参数的依赖类
class PageParams:
def __init__(self, page: int = 1, size: int = 10):
'''
解析分页参数
:param page: 当前页数
:param size: 每页的记录数
:return:
'''
self.page = page
self.size = size
# 在参数中指定默认值 Depends()
@app.get('/students')
async def get_student(pp: PageParams = Depends()):
print("page:", pp.page)
print("size:", pp.size)
return {
"code": 200,
"msg": "学习依赖"
}
测试
浏览器发请求：http://127.0.0.1:8000/students?page=2&size=30 ，控制台打印如下：
page: 2
size: 30
全局依赖（整个app所有接口生效）
所有路由都执行这个依赖，适合全局鉴权、日志：
async def global_dep():
print("每个请求进来都会执行")
app = FastAPI(dependencies=[Depends(global_dep)])
APIRouter级别依赖
只作用于这一组路由：
from fastapi import APIRouter
router = APIRouter(
prefix="/admin",
dependencies=[Depends(check_token)]
)
获取数据库会话（SQLAlchemy2.0）
async def get_db():
async with db_session() as session:
yield session
@app.get("/demo")
async def demo(session = Depends(get_db)):
pass
yield生成器依赖：执行路由前执行yield之前，请求结束后执行yield之后，适合关闭连接、资源释放。
Header获取token鉴权
from fastapi import Header, HTTPException
async def get_auth_token(authorization: str | None = Header(None)):
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="未授权")
return authorization.removeprefix("Bearer ")
@app.get("/profile")
async def profile(token: str = Depends(get_auth_token)):
return {"token": token}
总结
- 缓存：同一个请求中，同一个依赖只会执行一次，多处Depends同一个函数不会重复执行。
- 支持函数、类、生成器(yield)、async函数。
- 作用粒度：参数级别、路由级别、Router级别、全局App级别。
- 可以抛出
HTTPException中断请求。