跳到内容

依赖项

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

提示

如果可能,请优先使用 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

提示

如果可能,请优先使用 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

就是这样。

2行.

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

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

它可以返回任何您想要的东西。

在这种情况下,此依赖项需要

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

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

信息

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

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

请确保您将 FastAPI 版本升级 到至少 0.95.1 后再使用 Annotated

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

提示

如果可能,请优先使用 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

提示

如果可能,请优先使用 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

提示

如果可能,请优先使用 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

提示

如果可能,请优先使用 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()

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

提示

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

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

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

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

common_parameters --> read_items
common_parameters --> read_users

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

检查

请注意,您无需创建特殊类并将其传递给 FastAPI 以“注册”它或类似的东西。

您只需将其传递给 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

提示

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

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

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

当您在 大型代码库 中使用它时,这尤其有用,您会在 许多 路径操作 中反复使用 相同的依赖项

使用 async 还是不使用 async

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

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

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

这无关紧要。FastAPI 会知道该怎么做。

注意

如果您不清楚,请查看文档中关于 asyncawait异步:“紧急?” 部分。

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