跳转至

依赖项

FastAPI 拥有一个非常强大但直观的 依赖注入 系统。

它设计得非常易于使用,并且使得任何开发者都能轻松地将其他组件与 FastAPI 集成。

什么是“依赖注入”

“依赖注入” 在编程中意味着,你的代码(在这里是你的路径操作函数)有一种方式来声明它工作需要和使用的东西:“依赖项”。

然后,该系统(在这里是 FastAPI)将负责提供你的代码所需的这些依赖项(“注入”依赖项)。

这在以下情况下非常有用:

  • 需要共享逻辑(反复使用相同的代码逻辑)。
  • 共享数据库连接。
  • 强制执行安全性、身份验证、角色要求等。
  • 以及许多其他情况...

所有这些,同时最大限度地减少代码重复。

第一步

让我们看一个非常简单的例子。目前它非常简单,以至于不太有用。

但这样我们可以专注于依赖注入系统的工作原理。

创建一个依赖项,或“可依赖项”

让我们首先关注依赖项。

它只是一个可以接受与路径操作函数相同所有参数的函数:

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


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


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
🤓 Other versions and variants
from typing import Annotated, Union

from fastapi import Depends, FastAPI

app = FastAPI()


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


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


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


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = 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

Tip

Prefer to use the Annotated version if possible.

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = 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

就是这样。

两行代码

它具有与你所有路径操作函数相同的形状和结构。

你可以将其视为没有“装饰器”(没有 @app.get("/some-path"))的路径操作函数

并且它可以返回任何你想要的内容。

在这种情况下,这个依赖项期望:

  • 一个可选的查询参数 q,类型为 str
  • 一个可选的查询参数 skip,类型为 int,默认为 0
  • 一个可选的查询参数 limit,类型为 int,默认为 100

然后它只是返回一个包含这些值的 dict

Info

FastAPI 在 0.95.0 版本中添加了对 Annotated 的支持(并开始推荐使用)。

如果你使用的是旧版本,尝试使用 Annotated 时会出错。

在使用 Annotated 之前,请确保将 FastAPI 版本升级到至少 0.95.1。

导入 Depends

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


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


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
🤓 Other versions and variants
from typing import Annotated, Union

from fastapi import Depends, FastAPI

app = FastAPI()


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


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


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


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = 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

Tip

Prefer to use the Annotated version if possible.

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = 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

在“依赖项”中声明依赖

与你使用 BodyQuery 等处理路径操作函数参数的方式相同,使用 Depends 和一个新参数:

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


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


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
🤓 Other versions and variants
from typing import Annotated, Union

from fastapi import Depends, FastAPI

app = FastAPI()


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


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


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


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = 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

Tip

Prefer to use the Annotated version if possible.

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = 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

尽管你在函数参数中使用 Depends 的方式与使用 BodyQuery 等相同,但 Depends 的工作方式略有不同。

你只给 Depends 一个参数。

这个参数必须是一个类似函数的东西。

不直接调用它(不在末尾添加括号),只是将其作为参数传递给 Depends()

并且该函数接受参数的方式与路径操作函数相同。

Tip

你将在下一章看到除了函数之外,还有哪些其他“东西”可以用作依赖项。

每当有新请求到达时,FastAPI 将负责:

  • 使用正确的参数调用你的依赖(“可依赖”)函数。
  • 从你的函数获取结果。
  • 将该结果分配给你的路径操作函数中的参数。
graph TB

common_parameters(["common_parameters"])
read_items["/items/"]
read_users["/users/"]

common_parameters --> read_items
common_parameters --> read_users

这样你只需编写一次共享代码,FastAPI 负责为你的路径操作调用它。

Check

请注意,你不必创建一个特殊的类并将其传递到某处来“注册”它或类似的操作。

你只需将其传递给 DependsFastAPI 就知道如何完成其余的工作。

共享 Annotated 依赖项

在上面的示例中,你会看到存在一点代码重复

当你需要使用 common_parameters() 依赖项时,你必须编写带有类型注解和 Depends() 的整个参数:

commons: Annotated[dict, Depends(common_parameters)]

但是因为我们使用 Annotated,我们可以将该 Annotated 值存储在一个变量中,并在多个地方使用它:

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


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


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons
🤓 Other versions and variants
from typing import Annotated, Union

from fastapi import Depends, FastAPI

app = FastAPI()


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


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons
from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


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


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons

Tip

这只是标准的 Python,称为“类型别名”,实际上并非 FastAPI 特有。

但由于 FastAPI 基于 Python 标准,包括 Annotated,你可以在代码中使用这个技巧。😎

依赖项将继续按预期工作,最棒的部分类型信息将被保留,这意味着你的编辑器将能够继续为你提供自动补全内联错误等。其他工具如 mypy 也是如此。

当你在大型代码库中使用它,并且在许多路径操作中反复使用相同的依赖项时,这将特别有用。

使用 async 或不使用 async

由于依赖项也将由 FastAPI 调用(与你的路径操作函数相同),因此在定义函数时适用相同的规则。

你可以使用 async def 或普通的 def

你可以在普通 def 路径操作函数内部声明带有 async def 的依赖项,或在 async def 路径操作函数内部声明 def 依赖项,等等。

这没关系。FastAPI 会知道该怎么做。

Note

如果你不了解,请查看文档中关于 asyncawaitAsync: "In a hurry?" 部分。

与 OpenAPI 集成

你的依赖项(和子依赖项)的所有请求声明、验证和要求都将集成到同一个 OpenAPI 模式中。

因此,交互式文档也将包含这些依赖项的所有信息:

简单用法

如果你仔细想想,路径操作函数被声明为在路径操作匹配时使用,然后 FastAPI 负责使用正确的参数调用该函数,从请求中提取数据。

实际上,所有(或大多数)Web 框架都以相同的方式工作。

你从不直接调用这些函数。它们由你的框架(在这里是 FastAPI)调用。

通过依赖注入系统,你还可以告诉 FastAPI,你的路径操作函数还“依赖”于其他应在你的路径操作函数之前执行的东西,FastAPI 将负责执行它并“注入”结果。

其他用于描述相同“依赖注入”概念的常见术语有:

  • 资源
  • 提供者
  • 服务
  • 可注入对象
  • 组件

FastAPI 插件

可以使用依赖注入系统构建集成和“插件”。但实际上,实际上不需要创建“插件”,因为通过使用依赖项,可以声明无限数量的集成和交互,这些集成和交互对你的路径操作函数可用。

并且可以以非常简单和直观的方式创建依赖项,允许你只需导入所需的 Python 包,并用几行代码将它们与你的 API 函数集成,字面上

你将在接下来的章节中看到这方面的例子,关于关系数据库和 NoSQL 数据库、安全性等。

FastAPI 兼容性

依赖注入系统的简单性使得 FastAPI 兼容于:

  • 所有关系数据库
  • NoSQL 数据库
  • 外部包
  • 外部 API
  • 身份验证和授权系统
  • API 使用监控系统
  • 响应数据注入系统
  • 等等。

简单而强大

尽管分层依赖注入系统定义和使用起来非常简单,但它仍然非常强大。

你可以定义本身又可以定义依赖项的依赖项。

最后,构建了一个分层依赖树,依赖注入系统负责为你解决所有这些依赖项(及其子依赖项),并在每个步骤提供(注入)结果。

例如,假设你有 4 个 API 端点(路径操作):

  • /items/public/
  • /items/private/
  • /users/{user_id}/activate
  • /items/pro/

然后你可以仅使用依赖项和子依赖项为它们中的每一个添加不同的权限要求:

graph TB

current_user(["current_user"])
active_user(["active_user"])
admin_user(["admin_user"])
paying_user(["paying_user"])

public["/items/public/"]
private["/items/private/"]
activate_user["/users/{user_id}/activate"]
pro_items["/items/pro/"]

current_user --> active_user
active_user --> admin_user
active_user --> paying_user

current_user --> public
active_user --> private
admin_user --> activate_user
paying_user --> pro_items

OpenAPI 集成

所有这些依赖项,在声明它们的要求的同时,也向你的路径操作添加了参数、验证等。

FastAPI 将负责将其全部添加到 OpenAPI 模式中,以便它显示在交互式文档系统中。