依赖项¶
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
的 异步:“紧急?” 部分。
与 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 模式中,以便在交互式文档系统中显示。