fastapi 路径依赖项Depends / 装饰器依赖dependencies / 全局依赖 / 带 yield 的依赖

文章目录

- 1. 依赖项
- 2. 类作为依赖
- 3. 子依赖项
- - 3.1 多次使用同一个依赖项
- 4. 路径操作装饰器依赖项
- 5. 全局依赖项
- 6. 带 yield 的依赖项
- 7. 使用带 yield 上下文管理器作为依赖项

learn from https://fastapi.tiangolo.com/zh/tutorial/dependencies/

1. 依赖项

只能传给 Depends 一个参数。且该参数必须是可调用对象，比如函数

from typing import Optional
from fastapi import FastAPI, Dependsapp = FastAPI()# 依赖项函数，没有@app.xxx 
async def common_params(q: Optional[str]=None, skip: int = 0, limit: int = 100): return {"q":q, "skip":skip, "limit":limit}@app.get("/items/")
async def read_items(commons: dict = Depends(common_params)): # 传入依赖项return commons@app.get("/users/")
async def read_users(commons: dict = Depends(common_params)):return commons

在普通的 def 路径操作函数中，可以声明异步的 async def 依赖项
也可以在异步的 async def 路径操作函数中声明普通的 def 依赖项

在这里插入图片描述
交互式文档里也会显示 依赖的参数

2. 类作为依赖

只要可调用，就可以作为依赖项 callable
FastAPI 调用类，创建了一个实例，传给参数 commons

from typing import Optional
from fastapi import FastAPI, Dependsapp = FastAPI()fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]class CommonQueryParams:def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):self.q = qself.skip = skipself.limit = limit@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
# async def read_items(commons: CommonQueryParams = Depends()): # 也可以
# async def read_items(commons = Depends(CommonQueryParams)): # 也可以response = {}if commons.q:response.update({"q": commons.q})items = fake_items_db[commons.skip : commons.skip + commons.limit]response.update({"items": items})return response

在这里插入图片描述

3. 子依赖项

可以按需声明任意深度的子依赖项嵌套层级

from typing import Optional
from fastapi import FastAPI, Depends, Cookieapp = FastAPI()def query_extractor(q: Optional[str] = None):return qdef query_or_cookie_extractor(q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(None)
):if not q:return last_queryreturn q@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):return {"q_or_cookie": query_or_default}

两级嵌套
在这里插入图片描述

3.1 多次使用同一个依赖项

在同一个路径操作 多次声明了同一个依赖项，例如，多个依赖项共用一个子依赖项，FastAPI 在处理同一请求时，只调用一次该子依赖项，使用了缓存
如果不想使用「缓存」值，而是为需要在同一请求的每一步操作（多次）中都实际调用依赖项，可以把 Depends 的参数 use_cache 的值设置为 False

from typing import Optional
from fastapi import FastAPI, Depends, Cookieapp = FastAPI()def query_extractor(q: Optional[str] = None):print("run one time！")return qdef query_or_cookie_extractor(q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(None)
):print("run flag！")if not q:return last_queryreturn q@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor),another_query: str = Depends(query_extractor)):return {"q_or_cookie": query_or_default}

输出：

run one time！
run flag！

更改，不缓存

@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor),another_query: str = Depends(query_extractor, use_cache=False)):return {"q_or_cookie": query_or_default}

则输出：

run one time！
run flag！
run one time！

4. 路径操作装饰器依赖项

有时候，不需要依赖项的返回值，或者有的依赖项 不返回值，但仍要指向或解析该依赖项
可以在路径操作装饰器中添加一个由可选参数 dependencies 组成的 Depends() 的 list
就算这些依赖项会返回值，它们的值也不会传递给路径操作函数
可以触发异常

from fastapi import Depends, FastAPI, Header, HTTPExceptionapp = FastAPI()async def verify_token(x_token: str = Header(...)):if x_token != "fake-super-secret-token":raise HTTPException(status_code=400, detail="X-Token header invalid")async def verify_key(x_key: str = Header(...)):if x_key != "fake-super-secret-key":raise HTTPException(status_code=400, detail="X-Key header invalid")return x_key@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():return [{"item": "Foo"}, {"item": "Bar"}]

在这里插入图片描述

5. 全局依赖项

为 整个应用 添加依赖项，FastAPI(dependencies=[Depends(xxx), Depends(xx)])，所有的路径操作都依赖 dependencies 的内容

from fastapi import Depends, FastAPI, Header, HTTPExceptionasync def verify_token(x_token: str = Header(...)):if x_token != "fake-super-secret-token":raise HTTPException(status_code=400, detail="X-Token header invalid")async def verify_key(x_key: str = Header(...)):if x_key != "fake-super-secret-key":raise HTTPException(status_code=400, detail="X-Key header invalid")return x_keyapp = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])@app.get("/items/")
async def read_items():return [{"item": "Portal Gun"}, {"item": "Plumbus"}]@app.get("/users/")
async def read_users():return [{"username": "Rick"}, {"username": "Morty"}]

在这里插入图片描述

6. 带 yield 的依赖项

在依赖项结束的时候，做一些操作
如果需要，请在 yield 之前 raise 异常

async def get_db():db = DBSession()try:yield dbfinally:db.close()

7. 使用带 yield 上下文管理器作为依赖项

可以自己实现一个类，带 __enter__() ， __exit__() 函数的，可以作为上下文管理器

class MySuperContextManager:def __init__(self):self.db = DBSession()def __enter__(self):return self.dbdef __exit__(self, exc_type, exc_value, traceback):self.db.close()async def get_db():with MySuperContextManager() as db:yield db