依赖项¶

**FastAPI** 拥有一个非常强大但直观的 **依赖注入** 系统。

它被设计为非常易于使用，并且使任何开发人员都能轻松地将其他组件集成到 **FastAPI** 中。

什么是“依赖注入”¶

在编程中，**“依赖注入”** 表示你的代码（在本例中为你的路径操作函数）可以声明其工作和使用所需的事物：“依赖项”。

然后，该系统（在本例中为 **FastAPI**）将负责完成提供你的代码所需的这些依赖项（“注入”依赖项）所需的操作。

这在你需要时非常有用

拥有共享逻辑（一次又一次地使用相同的代码逻辑）。
共享数据库连接。
执行安全、身份验证、角色要求等。
以及其他许多事情...

所有这些，同时最大程度地减少代码重复。

第一步¶

让我们来看一个非常简单的例子。现在，它非常简单，所以没有太大用处。

但通过这种方式，我们可以专注于 **依赖注入** 系统的工作原理。

创建一个依赖项，或“可依赖的”¶

让我们先关注依赖项。

它只是一个函数，可以接受与路径操作函数相同的参数

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ 非 AnnotatedPython 3.8+ 非 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}


@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`¶

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ 非 AnnotatedPython 3.8+ 非 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}


@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

在“依赖者”中声明依赖项¶

就像你使用 Body、Query 等作为你的路径操作函数参数一样，使用 Depends 和一个新的参数

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ 非 AnnotatedPython 3.8+ 非 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}


@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 的方式与使用 Body、Query 等相同，但 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**以“注册”它或执行类似的操作。

您只需将其传递给Depends，**FastAPI**就会知道如何完成剩下的工作。

共享`Annotated` 依赖项¶

在上面的示例中，您可以看到存在一些**代码重复**。

当您需要使用common_parameters() 依赖项时，您必须使用类型注释和Depends()编写完整的参数。

commons: Annotated[dict, Depends(common_parameters)]

但由于我们使用的是Annotated，因此我们可以将该Annotated值存储在变量中，并在多个地方使用它。

Python 3.10+Python 3.9+Python 3.8+

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

由于依赖项也将由**FastAPI**调用（与您的路径操作函数相同），因此在定义函数时，相同的规则适用。

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

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

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

注意

如果您不知道，请查看文档中关于async和await的异步：“赶时间吗？”部分。

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

依赖项¶

什么是“依赖注入”¶

第一步¶

创建一个依赖项，或“可依赖的”¶

导入 Depends¶

在“依赖者”中声明依赖项¶

共享Annotated 依赖项¶

是否要使用async¶

与 OpenAPI 集成¶

简单用法¶

**FastAPI** 插件¶

**FastAPI** 兼容性¶

简单而强大¶

与**OpenAPI** 集成¶

导入 `Depends`¶

共享`Annotated` 依赖项¶

是否要使用`async`¶

FastAPI 插件¶

FastAPI 兼容性¶

与OpenAPI 集成¶