fastapi-Depends

FastAPI提供了简单易用，但功能强大的依赖注入系统。这个依赖系统设计的简单易用，可以让开发人员轻松地把组件集成至FastAPI。

编程中的「依赖注入」是声明代码（本文中为路径操作函数 ）运行所需的，或要使用的「依赖」的一种方式。然后，由系统（本文中为 FastAPI）负责执行任意需要的逻辑，为代码提供这些依赖（「注入」依赖项）。依赖注入常用于以下场景：

• 共享业务逻辑（复用相同的代码逻辑）
• 共享数据库连接
• 实现安全、验证、角色权限
• 等……

上述场景均可以使用依赖注入，将代码重复最小化

简单使用

要使用依赖项，我们就不得不使用到Depends函数，使用如下：

from typing import Dict

from fastapi import FastAPI
from fastapi import Query
from fastapi import Depends

app = FastAPI()

async def paging_params(per_page: int = Query(description='数量'),
                        page: int = Query(description='页码')) -> Dict[str, int]:
    return dict(per_page=per_page, page=page)

@app.get("/teachers")
async def teacher_list(params: dict = Depends(paging_params)):
    return params

@app.get("/students")
async def student_list(params: dict = Depends(paging_params)):
    return params
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

这样就避免了重复定义分页参数了，这样使得代码更加简洁！

在上面的例子中，我们是使用了一个函数作为依赖项。除此之外，我们可以使用一个类来作为依赖项，使用如下：

class PagingParams(object):

    def __init__(self, name: str = Query(description='名称')):
        self.name = name

@app.get("/teachers")
async def teacher_list(params: PagingParams = Depends()):
    return {'name': params.name}
1
2
3
4
5
6
7
8

所以只要它是一个Callable对象，那么它就可以作为一个依赖项使用！

有时，我们并不需要使用到依赖项的返回值，或者说有些依赖项不返回值，但是我们仍需要它执行或解析该依赖项。对于这种情况下，不必声明在视图函数的参数时使用Depends，而是可以在路径操作装饰器中添加一个由dependencies组成的list。如下：

from fastapi import Header

async def ip_allowed(x_forwarded_for: str = Header(description='ip地址')):
    if x_forwarded_for not in ['192.168.0.100']:
        raise ForbiddenError


@app.get("/teachers", dependencies=[Depends(ip_allowed)])
async def teacher_list():
    return {'code': 1}
1
2
3
4
5
6
7
8
9
10

路径装饰器依赖项的执行或解析方式和普通依赖项一样，但就算这些依赖项会返回值，它们的值也不会传递给路径操作函数。

嵌套依赖

FastAPI支持创建含子依赖项的依赖项。并且，可以按需声明任意深度的子依赖项嵌套层级，FastAPI负责处理解析不同深度的子依赖项。使用如下：

async def paging_params(per_page: int = Query(description='数量'),
                        page: int = Query(description='页码')) -> Dict[str, int]:
    return dict(per_page=per_page, page=page)


async def api_list_params(search: str = Query(description='查询参数'),
                          params: dict = Depends(paging_params)) -> Dict[str, int]:
    params['search'] = search
    return params


@app.get("/teachers")
async def teacher_list(params: dict = Depends(api_list_params)):
    return params
1
2
3
4
5
6
7
8
9
10
11
12
13
14

可以看到视图函数中的依赖项api_list_params，该函数又有依赖项paging_params，这就形成了一个嵌套依赖！

上面嵌套依赖的代码中，FastAPI必须先处理paging_params，以便在调用api_list_params时使用paging_params返回的结果！

全局依赖

有时，我们要为整个应用添加依赖项，那么我们又该如何实现呢？在上面已经讲到过路径装饰器的dependencies参数，而且我们知道FastAPI和APIRouter都是支持该参数的的，所以如果我们想要添加全局依赖，或者部分依赖的话，我们可以用到dependencies参数，如下：

from fastapi import Header
from fastapi import Depends
from fastapi import FastAPI

from app.utils import black_list


async def ip_allowed(x_forwarded_for: str = Header(description='ip地址')):
    if x_forwarded_for in black_list:   # 在黑名单中的IP禁止访问
        raise ForbiddenError


app = FastAPI(dependencies=[Depends(ip_allowed)])

@app.get("/teachers")
async def teacher_list():
    return {'code': 1}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

同理，如果我们只想给一组API添加依赖，我们可以在APIRouter中使用dependencies参数!!!

Dependencies with yield

FastAPI支持在完成后执行一些额外步骤的依赖项。为此，我们应该使用yield而不是return，然后在后边编写额外的步骤。使用如下：

async def get_db():
    db = Session()
    try:
        yield db
    finally:
        db.close()
1
2
3
4
5
6

注意事项

如果在同一个视图函数中多次声明了同一个依赖项，或者说多个依赖项共用一个子依赖项，FastAPI在处理同一请求时，只调用一次该子依赖项。如下：

from uuid import uuid4

async def get_uuid_strings() -> str:
    return uuid4().hex

@app.get("/home")
async def home(q1: str = Depends(get_uuid_strings),
               q2: str = Depends(get_uuid_strings)):
    return {'q1': q1, 'q2': q2}
1
2
3
4
5
6
7
8
9

当我们请求时，会发现q1与q2的值是一模一样的。这是因为FastAPI不会为同一个请求多次调用同一个依赖项，而是把依赖项的返回值进行缓存，并把它传递给同一请求中所有需要使用该返回值的依赖项！

在高级使用场景中，如果不想使用缓存，而是为需要在同一请求的每一步操作中都实际调用依赖项，可以把Depends的参数use_cache的值设置为False。如下：

@app.get("/home")
async def home(q1: str = Depends(get_uuid_strings, use_cache=False),
               q2: str = Depends(get_uuid_strings, use_cache=False)):
    return {'q1': q1, 'q2': q2}
1
2
3
4

如此，q1与q2的值将会是两个不一样的值了！