什么是依赖注入#

依赖注入是申明代码运行所需的,或要使用的依赖的一种方式,让后有系统负责执行任意需要的逻辑,为代码提供这些依赖,常用与

共享业务逻辑
共享链接
实现安全/角色权限/验证可以让代码重复最小化

1
from typing import Union
2

3
from fastapi import Depends, FastAPI
4

5
app = FastAPI()
6

7
async def common_parameters(
8
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
9
):
10
    return {"q": q, "skip": skip, "limit": limit}
11

12
@app.get("/items/")
13
async def read_items(commons: dict = Depends(common_parameters)):
14
    return commons
15

16
@app.get("/users/")
17
async def read_users(commons: dict = Depends(common_parameters)):
18
    return commons

本例中的依赖项预期接收如下参数：

类型为 str 的可选查询参数 q
类型为 int 的可选查询参数 skip，默认值是 0
类型为 int 的可选查询参数 limit，默认值是 100

然后，依赖项函数返回包含这些值的 dict。

类作为依赖项#

fastapi 的依赖项只要是可调用对象就都可以

1
from fastapi import Depends, FastAPI
2

3
app = FastAPI()
4

5
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
6

7
class CommonQueryParams:
8
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
9
        self.q = q
10
        self.skip = skip
11
        self.limit = limit
12

13
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
14
    return {"q": q, "skip": skip, "limit": limit}
15

16
@app.get("/items/")
17
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
18
    response = {}
19
    if commons.q:
20
        response.update({"q": commons.q})
21
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
22
    response.update({"items": items})
23
    return response

commons: CommonQueryParams = Depends() 这样可以快捷的写 ()

子依赖项#

Fastapi 支持创建包含子依赖的依赖项

1
from typing import Union
2

3
from fastapi import Cookie, Depends, FastAPI
4

5
app = FastAPI()
6

7
def query_extractor(q: Union[str, None] = None):
8
    return q
9

10
def query_or_cookie_extractor(
11
    q: str = Depends(query_extractor),
12
    last_query: Union[str, None] = Cookie(default=None),
13
):
14
    if not q:
15
        return last_query
16
    return q
17

18
@app.get("/items/")
19
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
20
    return {"q_or_cookie": query_or_default}

路径操作依赖项#

有时候我们并不需要在路径操作函数中使用依赖项的放回值,但是依然需要解析或执行该依赖项,我们就可以在路径操作装饰器中添加

1
from fastapi import Depends, FastAPI, Header, HTTPException
2

3
app = FastAPI()
4

5
async def verify_token(x_token: str = Header()):
6
    if x_token != "fake-super-secret-token":
7
        raise HTTPException(status_code=400, detail="X-Token header invalid")
8

9
async def verify_key(x_key: str = Header()):
10
    if x_key != "fake-super-secret-key":
11
        raise HTTPException(status_code=400, detail="X-Key header invalid")
12
    return x_key
13

14
@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
15
async def read_items():
16
    return [{"item": "Foo"}, {"item": "Bar"}]