跳到内容

依赖项

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
🤓 其他版本和变体

提示

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

就是这样。

2 行代码.

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

你可以把它想象成一个没有“装饰器”(没有 @app.get("/some-path"))的路径操作函数

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

在这个例子中,此依赖项期望:

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

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

信息

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
🤓 其他版本和变体

提示

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

在“依赖方”中声明依赖项

就像在路径操作函数参数中使用 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
🤓 其他版本和变体

提示

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

尽管你在函数参数中使用 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

提示

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

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

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

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

是否使用 async

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

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

你可以在普通的 def 路径操作函数中声明 async def 依赖项,或者在 async def 路径操作函数中声明 def 依赖项,依此类推。

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

注意

如果你不确定,请查看文档中关于 asyncawaitAsync: “赶时间吗?” 部分。

与 OpenAPI 集成

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

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

简单用法

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

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

你从不直接调用这些函数。它们是由你的框架(本例中为 FastAPI)调用的。

借助依赖注入系统,你还可以告诉 FastAPI,你的路径操作函数还“依赖”于其他某些需要在你的路径操作函数之前执行的内容,FastAPI 将负责执行它并“注入”结果。

对于这种“依赖注入”概念,其他常见的术语包括:

  • 资源 (resources)
  • 提供程序 (providers)
  • 服务 (services)
  • 可注入对象 (injectables)
  • 组件 (components)

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