Skip to content

依赖项

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

如果可能,优先使用 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: dict = Depends(common_parameters)):
    return commons


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

Tip

如果可能,优先使用 Annotated 版本。

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

如果可能,优先使用 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: dict = Depends(common_parameters)):
    return commons


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

Tip

如果可能,优先使用 Annotated 版本。

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

如果可能,优先使用 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: dict = Depends(common_parameters)):
    return commons


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

Tip

如果可能,优先使用 Annotated 版本。

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
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 模式中,以便在交互式文档系统中显示。