跳到内容

依赖项

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 基于包括 Annotated 在内的 Python 标准,你可以在你的代码中使用这个技巧。😎

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

当你在一个大型代码库中,在许多路径操作中一遍又一遍地使用相同的依赖项时,这将特别有用。

使用 async 还是不使用 async

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

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

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

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

注意

如果你不确定,请查看文档中关于 asyncawait异步:“没时间解释了?” 部分。

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