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

文章目录


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 依赖项
  • fastapi 路径依赖项Depends / 装饰器依赖dependencies / 全局依赖 / 带 yield 的依赖


交互式文档里也会显示 依赖的参数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 内置默认会去处理。
上一篇:漫画:什么是桶排序?


下一篇:谨慎修改网站关键词,会影响排名