依赖项¶
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
Tip
如果可能,优先使用 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
Tip
如果可能,优先使用 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
就是这样。
两行代码。
它具有与所有 路径操作函数 相同的形状和结构。
你可以把它看作是一个没有“装饰器”(没有 @app.get("/some-path")
)的 路径操作函数。
并且它可以返回你想要的任何内容。
在这个例子中,这个依赖项期望:
- 一个可选的查询参数
q
,类型为str
。 - 一个可选的查询参数
skip
,类型为int
,默认值为0
。 - 一个可选的查询参数
limit
,类型为int
,默认值为100
。
然后它只返回一个包含这些值的 dict
。
Info
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
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
Tip
如果可能,优先使用 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
Tip
如果可能,优先使用 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
并添加一个新参数:
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
Tip
如果可能,优先使用 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
Tip
如果可能,优先使用 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()
。
而这个函数接受参数的方式与*路径操作函数*相同。
Tip
你将在下一章中看到除了函数之外,还有哪些“东西”可以用作依赖项。
每当有新的请求到达时,FastAPI 会负责:
- 使用正确的参数调用你的依赖项(“可依赖”)函数。
- 获取你函数的结果。
- 将该结果分配给你的*路径操作函数*中的参数。
graph TB
common_parameters(["common_parameters"])
read_items["/items/"]
read_users["/users/"]
common_parameters --> read_items
common_parameters --> read_users
通过这种方式,你只需编写一次共享代码,FastAPI 会负责为你的*路径操作*调用它。
Check
注意,你不需要创建一个特殊的类并将其传递到某个地方以“注册”它或类似的操作。
你只需将其传递给 Depends
,FastAPI 就知道如何处理其余的事情。
共享 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
Tip
这只是标准的 Python,它被称为“类型别名”,实际上并不特定于 FastAPI。
但由于 FastAPI 基于 Python 标准,包括 Annotated
,你可以在你的代码中使用这个技巧。😎
依赖项将继续按预期工作,而**最好的部分**是**类型信息将被保留**,这意味着你的编辑器将继续为你提供**自动补全**、**内联错误**等功能。对于 mypy
等其他工具也是如此。
当你在**大型代码库**中使用**相同的依赖项**在**许多*路径操作***中重复使用时,这将特别有用。
使用 async
还是不使用 async
¶
由于依赖项也将由 FastAPI 调用(与你的*路径操作函数*相同),因此在定义函数时适用相同的规则。
你可以使用 async def
或普通的 def
。
你可以在普通的 def
*路径操作函数*中声明带有 async def
的依赖项,或者在 async def
*路径操作函数*中声明带有 def
的依赖项,等等。
这并不重要。FastAPI 会知道该怎么做。
Note
如果你不确定,请查看文档中关于 async
和 await
的 Async: "In a hurry?" 部分。
与 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 模式中,以便在交互式文档系统中显示。