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

Michael阿明

发布于 2022-01-07 11:34:42

2.9K00

代码可运行

文章被收录于专栏：Michael阿明学习之路Michael阿明学习之路

运行总次数：0

代码可运行

文章目录
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, Depends

app = 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, Depends

app = 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 = q
        self.skip = skip
        self.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, Cookie

app = FastAPI()

def query_extractor(q: Optional[str] = None):
    return q


def query_or_cookie_extractor(
    q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(None)
):
    if not q:
        return last_query
    return 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, Cookie

app = FastAPI()

def query_extractor(q: Optional[str] = None):
    print("run one time！")
    return q


def query_or_cookie_extractor(
    q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(None)
):
    print("run flag！")
    if not q:
        return last_query
    return 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, HTTPException

app = 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, HTTPException


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 = 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 db
    finally:
        db.close()

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

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

class MySuperContextManager:
    def __init__(self):
        self.db = DBSession()
    def __enter__(self):
        return self.db
    def __exit__(self, exc_type, exc_value, traceback):
        self.db.close()

async def get_db():
    with MySuperContextManager() as db:
        yield db

注意： https://docs.python.org/3/library/contextlib.html#contextlib.contextmanager