前几篇我们把接口的输入、输出、错误处理都过了一遍。这一篇来看另一个问题:多个接口都要做同一件事,比如都要校验请求头里的 token,难道每个接口都复制粘贴一遍吗?
做完这篇后,我们会有两个接口共用同一段校验逻辑,接口函数本身只关心自己的业务,公共逻辑被拿到外面去了。
先看一个具体的问题
假设现在有个需求:所有接口都要在校验请求头里带一个x-token,值必须是secret-token。如果不是,返回 401。
最直接的写法是这样的:
fromfastapiimportHeader,HTTPException@app.get("/items/{item_id}",response_model=ItemPublic)defread_item(item_id:int,x_token:str=Header(...)):ifx_token!="secret-token":raiseHTTPException(status_code=401,detail="Invalid token")ifitem_idnotinfake_items_db:raiseHTTPException(status_code=404,detail="Item not found")item=fake_items_db[item_id]return{"id":item_id,**item}@app.get("/users/{user_id}")defread_user(user_id:int,x_token:str=Header(...)):ifx_token!="secret-token":raiseHTTPException(status_code=401,detail="Invalid token")ifuser_idnotinfake_users_db:raiseHTTPException(status_code=404,detail="User not found")return{"id":user_id,**fake_users_db[user_id]}重点看if x_token != "secret-token"这一段。它出现在两个接口里,一模一样。
如果再加十个接口,就要复制粘贴十次。哪天 token 的校验规则变了,比如要改成从数据库查,就要改十个地方。
这不是技术问题,是维护问题。
把公共逻辑拿出去
FastAPI 的依赖注入就是为了解决这个问题。它不是什么玄学,就是把一个函数的结果"注入"到另一个函数里。
先创建一个dependencies.py:
fromfastapiimportHeader,HTTPExceptiondefget_token_header(x_token:str=Header(...)):ifx_token!="secret-token":raiseHTTPException(status_code=401,detail="Invalid token")returnx_token这个函数只做一件事:检查x-token,不通过就抛异常,通过就把 token 返回。
然后改写接口:
fromfastapiimportDependsfrom.dependenciesimportget_token_header@app.get("/items/{item_id}",response_model=ItemPublic)defread_item(item_id:int,token:str=Depends(get_token_header)):ifitem_idnotinfake_items_db:raiseHTTPException(status_code=404,detail="Item not found")item=fake_items_db[item_id]return{"id":item_id,**item}@app.get("/users/{user_id}")defread_user(user_id:int,token:str=Depends(get_token_header)):ifuser_idnotinfake_users_db:raiseHTTPException(status_code=404,detail="User not found")return{"id":user_id,**fake_users_db[user_id]}重点看token: str = Depends(get_token_header)这一行。Depends(get_token_header)告诉 FastAPI:在调用read_item之前,先调用get_token_header,把它的返回值赋给token参数。
如果get_token_header抛了异常,FastAPI 不会进read_item,直接返回错误响应。
在/docs里试一次
启动服务:
.\.venv\Scripts\Activate.ps1$env:PYTHONIOENCODING ="utf-8"$env:PYTHONUTF8 ="1"fastapi dev app/main.py打开:
http://127.0.0.1:8000/docs点开GET /items/{item_id},会看到多了一个x-token的请求头输入框。这是 FastAPI 从get_token_header的参数定义里自动提取出来的,不需要我们手动写文档。
先不填x-token,直接点Execute。应该能看到 401 响应:
{"detail":"Invalid token"}填x-token为secret-token,再点Execute。这次应该能看到正常的 200 响应。
再试GET /users/{user_id},也是一样的行为。两个接口共用同一段校验逻辑,但校验代码只写了一遍。
FastAPI 在背后帮你做了什么
Depends做的事情很简单:在调用你的接口函数之前,先调用你指定的依赖函数,把依赖函数的返回值作为参数传给你的接口函数。
用普通 Python 代码类比,大概是这样的:
# FastAPI 帮你做的事情,类似这样:defhandle_request():# 1. 先调用依赖token=get_token_header(x_token=extract_header("x-token"))# 2. 再把依赖的返回值传给接口函数response=read_item(item_id=1,token=token)returnresponse只不过 FastAPI 是自动做的,包括从请求里提取参数、处理校验错误、生成文档,都是自动的。
这也是为什么get_token_header的函数签名里可以直接写x_token: str = Header(...)。FastAPI 在调用依赖函数时,会用和调用接口函数一样的方式解析参数。
依赖函数也可以不返回值
前面的get_token_header返回了x_token,但有些依赖只需要做校验,不需要把结果传给接口函数。
比如,只需要校验 token 存在且合法,接口函数不需要知道 token 的具体值:
fromfastapiimportHeader,HTTPExceptiondefverify_token(x_token:str=Header(...)):ifx_token!="secret-token":raiseHTTPException(status_code=401,detail="Invalid token")# 不 return 任何东西@app.get("/items/{item_id}",response_model=ItemPublic)defread_item(item_id:int,_:None=Depends(verify_token)):ifitem_idnotinfake_items_db:raiseHTTPException(status_code=404,detail="Item not found")item=fake_items_db[item_id]return{"id":item_id,**item}这里Depends(verify_token)的返回值是None,接口函数用_接收。FastAPI 仍然会先调用verify_token,只是接口函数里用不到它的返回值。
真实项目里常见的依赖场景
除了校验 token,真实项目里依赖注入还常用来做这些事:
分页参数。很多列表接口都需要skip和limit,每个接口都写一遍很烦:
fromtypingimportAnnotatedfromfastapiimportQuerydefget_pagination(skip:int=Query(0,ge=0),limit:int=Query(10,le=100)):return{"skip":skip,"limit":limit}@app.get("/items")deflist_items(pagination:dict=Depends(get_pagination)):skip=pagination["skip"]limit=pagination["limit"]...数据库连接。每个需要查数据库的接口都先要拿到一个数据库 session,用完后关闭:
defget_db():db=SessionLocal()try:yielddbfinally:db.close()@app.get("/items/{item_id}")defread_item(item_id:int,db:Session=Depends(get_db)):item=db.query(Item).filter(Item.id==item_id).first()...注意这里用了yield而不是return。FastAPI 支持这种写法:在yield之后的代码会在请求处理完后执行,适合做清理工作。
动手改一下
新增一个get_common_query依赖,统一读取q和limit两个查询参数,让两个接口都复用它。
先在dependencies.py里加:
fromtypingimportAnnotatedfromfastapiimportQuerydefget_common_query(q:str|None=Query(None,description="搜索关键词"),limit:int=Query(10,ge=1,le=100,description="返回数量上限"),):return{"q":q,"limit":limit}然后给GET /items和GET /users加上这个依赖(如果还没有列表接口,先加一个):
@app.get("/items")deflist_items(queries:dict=Depends(get_common_query)):q=queries["q"]limit=queries["limit"]# 简单模拟:如果有 q,只返回名字匹配的result=[]foritem_id,iteminfake_items_db.items():ifqisNoneorq.lower()initem["name"].lower():result.append({"id":item_id,**item})returnresult[:limit]保存后在/docs里调用GET /items,应该能看到q和limit两个查询参数仍然出现在文档里,和直接写在接口函数参数里时一样。
如果能在文档里看到这两个参数,并且传不同的值能得到不同的过滤结果,这篇就算真正学完。
到这里,这篇的目标已经完成:
- 我们知道了依赖注入是把公共逻辑从接口函数里拿出去的一种方式,不是玄学。
- 我们跑通了
get_token_header被两个接口复用的效果。 - 我们能在
/docs里看到依赖函数的参数仍然正常出现在文档里。
本文代码:https://github.com/tanghaojin/fastapi-beginner-lab/tree/article-06-dependencies
下一篇解决另一个问题:接口多了main.py写不下时,用APIRouter拆成多个文件。
参考资料
- FastAPI Dependencies: https://fastapi.tiangolo.com/tutorial/dependencies/