依赖项¶
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
在“依赖者”中声明依赖项¶
就像你在路径操作函数参数中使用 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
提示
如果可能,请优先使用 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
值存储在一个变量中并在多个地方使用它
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 会知道如何处理。
注意
如果你不清楚,请查看文档中关于 async
和 await
的 Async: “赶时间?” 部分。
与 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 模式中,以便在交互式文档系统中显示。