跳到内容

依赖项

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 会知道如何处理。

注意

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

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