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

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

假设有一个类 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 应用：

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