在 FastAPI 中,Depends 是一个依赖注入系统,用于注入应用程序中所需的依赖项,通过 Depends,我们可以轻松地将依赖项注入到 FastAPI 路由函数中。简单来说,Depends 依赖注入的目的就是将代码重复最小!
目录
1 Depends 使用
1.1 依赖注入介绍
1.2 demo
1.3 OpenAPI 集成
2 不同依赖项
2.1 类作为依赖项
① 可调用对象
② Python 类依赖注入
2.2 嵌套依赖项
2.3 dependencies 参数
2.4 全局依赖项
📌源码地址:
FastAPI_Study_Yinyu: FastAPI学习路径,CSDN专栏:http://t.csdn.cn/JRtrk
1 Depends 使用
FastAPI 提供了简单易用,但功能强大的依赖注入系统,这可以让开发人员轻松地把组件集成至 FastAPI。
1.1 依赖注入介绍
依赖注入是一种设计模式,用于降低程序组件之间的耦合度。它通过将组件之间的依赖关系从代码中分离出来,使得组件可以更加灵活地被替换、修改或重用。
假设有一个类 A 需要使用类B的功能,如果在 A 类中直接实例化 B 类,那么 A 类和 B 类之间的依赖关系就会很紧密,难以达到解耦的目的。而使用依赖注入的方式,A 类不再直接实例化 B 类,而是将 B 类的实例通过构造函数、属性、或者接口等方式注入到 A 类中。
依赖注入常用于以下场景:
- 共享业务逻辑(复用相同的代码逻辑)
- 共享数据库连接
- 实现安全、验证、角色权限
- 等……
上述场景均可以使用依赖注入,将代码重复最小化。
1.2 demo
这是一个非常简单的例子,通过这个例子,您可以初步了解「依赖注入」的工作机制。
from typing import Union
from fastapi import Depends, FastAPI #2导入 Depends
app = FastAPI()
#1创建依赖项
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)): #3声明依赖项
return commons
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)): #3声明依赖项
return commons
此时访问 127.0.0.1:8000/users/?q=yinyu&skip=1&limit=10 ,响应正常:
📌依赖项
依赖项 common_parameters 预期接收如下参数:
- 类型为 str 的可选查询参数 q
- 类型为 int 的可选查询参数 skip,默认值是 0
- 类型为 int 的可选查询参数 limit,默认值是 100
然后,依赖项函数将返回包含这些值的 dict。
声明依赖项时,这里只能传给 Depends 一个参数,且该参数必须是可调用对象。比如函数,该函数接收的参数等于路径操作函数的参数。
📌实际请求时,FastAPI 执行如下操作:
- 用正确的参数调用依赖项函数(「可依赖项」)
- 获取函数返回的结果
- 把函数返回的结果赋值给路径操作函数的请求参数,例如 commons
如此,只编写一次代码,FastAPI 就可以为多个路径操作/请求接口共享这段代码 。
注意,无需创建专门的类,并将之传递给 FastAPI 以进行「注册」或执行类似的操作。 只要把它传递给 Depends,FastAPI 就知道该如何执行后续操作。
1.3 OpenAPI 集成
依赖项及子依赖项的所有请求声明、验证和需求都可以集成至同一个 OpenAPI 概图。
因此,交互文档里也会显示依赖项的所有信息:
依赖注入系统如此简洁的特性,让 FastAPI 可以与下列系统兼容:
- 关系型数据库
- NoSQL 数据库
- 外部支持库
- 外部 API
- 认证和鉴权系统
- API 使用监控系统
- 响应数据注入系统
- 等等……
2 不同依赖项
在前面的例子中, 我们从依赖项中返回了一个 dict,但是我们知道编辑器不能为 dict 提供很多支持(比如补全),因为编辑器不知道 dict 的键和值类型, 因此接下来将介绍更好用的方式~
依赖项不只是函数,只要是 "可调用对象" 就可以,Python 中的 "可调用对象" 是指任何 Python 可以像函数一样 "调用" 的对象。
2.1 类作为依赖项
① 可调用对象
创建一个 Python 类的实例,您可以使用和依赖注入相同的语法,比如:
class Cat:
def __init__(self, name: str):
self.name = name
fluffy = Cat(name="Mr Fluffy")
在这个例子中,fluffy 是一个 Cat 类的实例。 为了创建 fluffy,调用了 Cat 。 所以,Python 类也是 可调用对象。
因此,在 FastAPI 中,你可以使用一个 Python 类作为一个依赖项。
② Python 类依赖注入
因此我们可以将 demo 里的依赖项 common_parameters 更改为类 CommonQueryParams:
class CommonQueryParams:
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
它与我们 demo 里的 common_parameters 具有相同的参数。
📌声明依赖项
@app.get("/items2/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
return commons
FastAPI 调用 CommonQueryParams 类。这将创建该类的一个 "实例",该实例将作为参数 commons 被传递给你的函数。
📌简化
注意,我们在上面的代码中编写了两次 CommonQueryParams:
commons: CommonQueryParams = Depends(CommonQueryParams)
FastAPI 对此做了简化,你可以写成:
commons=Depends(CommonQueryParams) 或 commons: CommonQueryParams = Depends()
📌完整代码
from typing import Union
from fastapi import Depends, FastAPI
app = FastAPI()
class CommonQueryParams:
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items2/")
async def read_items(commons = Depends(CommonQueryParams)):
return commons
此时访问 127.0.0.1:8000/items2/?q=yinyu&skip=2&limit=10 ,响应正常:
2.2 嵌套依赖项
FastAPI 支持创建含子依赖项的依赖项,并且可以按需声明任意深度的子依赖项嵌套层级。
📌 第一层依赖项
def query_extractor(q: Union[str, None] = None):
return q
这段代码声明了类型为 str 的可选查询参数 q,然后返回这个查询参数, 这个函数很简单。
📌 第二层依赖项
接下来,创建另一个依赖项函数,并同时用该依赖项自身再声明一个依赖项:
def query_or_cookie_extractor(
q: str = Depends(query_extractor),
last_query: Union[str, None] = Cookie(default=None),
):
if not q:
return last_query
return q
这里重点说明一下声明的参数:
- 尽管该函数自身是依赖项,但还声明了另一个依赖项
- 该函数依赖 query_extractor, 并把 query_extractor 的返回值赋给参数 q
- 同时,该函数还声明了类型是 str 的可选 cookie
- 用户未提供查询参数 q 时,则使用上次使用后保存在 cookie 中的查询
📌 使用嵌套依赖项
...
def query_extractor(q: Union[str, None] = None):
return q
def query_or_cookie_extractor(
q: str = Depends(query_extractor),
last_query: Union[str, None] = Cookie(default=None),
):
if not q:
return last_query
return q
@app.get("/items3/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
return {"q_or_cookie": query_or_default}
注意,这里在路径操作函数中只声明了一个依赖项,即 query_or_cookie_extractor 。
但 FastAPI 必须先处理 query_extractor,以便在调用 query_or_cookie_extractor 时使用 query_extractor 返回的结果。类似下边的路径图:
此时访问 127.0.0.1:8000/items3/?q=yinyu ,响应正常:
📌 依赖项缓存
如果多次声明了同一个依赖项,例如多个依赖项共用一个子依赖项,那么 FastAPI 在处理同一请求时,只调用一次该子依赖项。
FastAPI 默认不会为同一个请求多次调用同一个依赖项,而是把依赖项的返回值进行「缓存」,并把它传递给同一请求中所有需要使用该返回值的「依赖项」。
如果不想使用「缓存」值,而是为需要在同一请求的每一步操作(多次)中都实际调用依赖项,可以把 Depends 的参数 use_cache 的值设置为 False :
async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):
return {"fresh_value": fresh_value}
2.3 dependencies 参数
有时,我们并不需要在路径操作函数中使用依赖项的返回值,有些依赖项也不返回值,或者说接口根本用不上这些返回值~
对于这种情况,不必在声明路径操作函数的参数时使用 Depends,而是可以在路径操作装饰器中添加一个由 dependencies 组成的 list。
路径操作装饰器支持可选参数 ~ dependencies, 该参数的值是由 Depends() 组成的 list:
from fastapi import Depends, FastAPI, Header, HTTPException
app = FastAPI()
#路径装饰器依赖项可以声明请求的需求项(比如响应头)或其他子依赖项
async def verify_token(x_token: str = Header()):
if x_token != "fake-super-secret-token":
#路径装饰器依赖项与正常的依赖项一样,可以 raise 异常:
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def verify_key(x_key: str = Header()):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
@app.get("/items4/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
return [{"item": "Foo"}, {"item": "Bar"}]
路径操作装饰器依赖项(以下简称为“路径装饰器依赖项”)的执行或解析方式和普通依赖项一样,但就算这些依赖项会返回值,它们的值也不会传递给路径操作函数。
2.4 全局依赖项
有时,我们要为整个应用添加依赖项。 通过与定义路径装饰器依赖项类似的方式,可以把依赖项添加至整个 FastAPI 应用:文章来源:https://www.toymoban.com/news/detail-474710.html
from fastapi import Depends, FastAPI, Header, HTTPException
async def verify_token(x_token: str = Header()):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def verify_key(x_key: str = Header()):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
app1 = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
@app1.get("/items5/")
async def read_items():
return [{"item": "Portal Gun"}, {"item": "Plumbus"}]
那么所有 app 请求都将经过 verify_token 和 verify_key 函数,类似于过滤器。文章来源地址https://www.toymoban.com/news/detail-474710.html
到了这里,关于【Python开发】FastAPI 07:Depends 依赖注入的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!