fastapi教程-进阶九（Dependencies-1）

参考内容：

• https://fastapi.tiangolo.com/

fastapi有非常强大的依赖注入系统，虽然听起来会很难，但是用起来非常简单，并且非常方便我们集成各种组件

什么是依赖注入

依赖注入是指在我们的项目中，定义了一些方法，这些方法是我们某些路径方法需要依赖的，这些方法叫做依赖项，当代码运行时，fastapi会将这个依赖项注入到路径方法中。

如果很难理解，我们可以通过了解他的使用场景来理解他的意思。依赖注入可以用在下面的场景：

• 有共享逻辑时，也就是重复的业务逻辑
• 共享数据连接
• 安全机制、权限校验、角色管理等等
• 其他的可以复用或重复的逻辑

通过使用依赖注入，可以大大减少重复代码。

快速入门

我们先来看一个简单的例子，从而理解fastapi的依赖注入：

from typing import Optional

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(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_parameters)):
    return commons


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

在这里个例子中，我们可以发现read_items和read_users这两个路径方法都使用了common_parameters这个依赖项，这个依赖项希望得到：

• string类型的q参数
• int类型的，默认值是0的skip参数
• int类型，默认值是0的limit参数

并且，这个依赖项会返回一个由参数组成的dict

需要注意的一点，common_parameters是一个方法，是一个可调用对象，也就意味着Depends需要接收一个可调用对象作为参数

我们先来启动服务，打开http://127.0.0.1:8000/docs，看一看她会生成怎样的api：

不难发现，api接收的参数就是依赖项的参数，到这里，我们尝试一下请求这两个api，其实我们可以分析出来当一个请求进来时fastapi是如何使用依赖注入的了：

1. 先调用依赖项这个方法，并传入正确的参数
2. 执行依赖项，得到返回
3. 将依赖项返回结果指派给路径方法对应的参数

如果很难理解，可以看下面这个图：

common_parameters

/items/

/users/

通过这个入门可以快速学习到如何使用依赖注入，我们可以得出一个结论：在使用依赖注入时，我们不需要做其他的事情，只需要将依赖项定义好，并告诉fastapi哪些路径方法需要使用这个依赖项，其他的都交给fastapi就好，当请求进来后，fastapi会负责将依赖项注入到方法中。

Fastapi兼容性

依赖注入的简单性使得fastapi具备了更好的兼容性，通过依赖注入fastapi可以兼容：

• 所有关系数据库
• NoSQL数据库
• 外部包装
• 外部API
• 认证和授权系统
• API使用情况监控系统
• 响应数据注入系统
• 等等

用类做依赖项

在上面的快速入门中，我们学习了如何用方法来做依赖项，我们也提到了，Depends需要接收一个可调用对象作为参数，那么也就是说，依赖项必须是可调用对象。下面我们尝试用类来做依赖项：

from typing import Optional

from fastapi import Depends, FastAPI

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)):
    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

注意看CommonQueryParams类的__init__方法，他接收的参数和上面的common_parameters方法是一样的：

def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):

然后我们告诉fastapi接口的依赖项是CommonQueryParams类:

async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):

当请求进来后，fastapi会调用CommonQueryParams类，并创建一个实例，将这个实例赋值给commons参数。

depends简化写法

在上面的例子中，我们注意到CommonQueryParams写了两遍:

commons: CommonQueryParams = Depends(CommonQueryParams)

其实，第一个CommonQueryParams并没有什么特殊含义，fastapi也不会用它来做数据验证，而第二个CommonQueryParams才是fastapi会用到的依赖项，因此，我们可以简化成：

commons=Depends(CommonQueryParams)

此时的代码就变成了：

from typing import Optional

from fastapi import Depends, FastAPI

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=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也鼓励我们用commons: CommonQueryParams = Depends(CommonQueryParams)这样的形式来声明参数类型，这样写我们的编译器可以知道这个参数的类型，从而帮助我们进行代码补全或检查:

那么如何同时具备这两个特点呢，我们可以这样写：

commons: CommonQueryParams = Depends()

此时的代码变成了：

from typing import Optional

from fastapi import Depends, FastAPI

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()):
    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

上述栗子均放到git上啦，地址：戳这里