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

2022-03-24 07:04:20

文章目录

1. 依赖项

2. 类作为依赖

3. 子依赖项

3.1 多次使用同一个依赖项

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

5. 全局依赖项

6. 带 yield 的依赖项

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

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. 类作为依赖只要可调用，就可以作为依赖项 callableFastAPI 调用类，创建了一个实例，传给参数 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

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

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}

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

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}

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

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"}]

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

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"}]

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

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

注意：

使用 @contextlib.contextmanager or @contextlib.asynccontextmanager 装饰带一个yield 的函数，也可以创建上下文管理
但是你不能写 @xxx， FastAPI 内置默认会去处理。

码农公寓

3. 子依赖项

5. 全局依赖项

6. 带 yield 的依赖项

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

相关文章