【Python开发】FastAPI 07:Depends 依赖注入

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 依赖注入介绍

依赖注入是一种设计模式,用于降低程序组件之间的耦合度。它通过将组件之间的依赖关系从代码中分离出来,使得组件可以更加灵活地被替换、修改或重用。

假设有一个类 需要使用类B的功能,如果在 类中直接实例化 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 以进行「注册」或执行类似的操作。 只要把它传递给 DependsFastAPI 就知道该如何执行后续操作。

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 应用:

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 函数,类似于过滤器。

本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
THE END
分享
二维码
< <上一篇
下一篇>>