FastAPI 依赖注入详解：生成依赖树

Posted 2023-03-23

参考技术A

在添加 APIRoute 节点时，会对endpoint进行解析，生成 依赖树 ， get_dependant 便是解析出endpoint的依赖树的函数。

这部分在之前源码解析中讲过，但是当时的理解并不深刻。这次让我们来认真剖析这部分

get_dependant 不止被endpoint使用，其依赖项和子依赖都会使用，其为递归函数。
开头生成一个 Dependant节点对象 ，等待下面加工，最终被返回。其形成的是一个 树状结构 。

接下来把该节点的参数都抓出来，逐个分析。

首先判断是否为 Depends() 项，如果是，则生成子依赖。下面是生成子依赖的流程。

拿出Depends中的依赖内容，如果没有就用注解来充当。即 user: User = Depends() 这种形式可以被允许。

接下来是对安全相关的处理。我们可以看到，中间又调用了 get_dependant ，参数包含了 name 和 security_scopes 。endpoint的根节点传参不包含这两项。

如果不是 Depends 参数，则首先默认当成查询参数query，并生成ModelField字段。

如果其为路径参数，则重新生成ModelField字段。再整合到dependant的参数列表中

不是路径参数，但是标准的查询参数

Query()和Header()两种情况

当上述条件都不满足，则可以断言为Body()字段。

就此，一个APIRoute的依赖树便生成了
下章说说如何使用依赖树

FastAPI Web框架 [依赖项]

依赖项

FastAPI 提供了简单易用，但功能强大的依赖注入系统。这个依赖系统设计的简单易用，可以让开发人员轻松地把组件集成至 FastAPI。

编程中的「依赖注入」是声明代码（本文中为路径操作函数 ）运行所需的，或要使用的「依赖」的一种方式。
然后，由系统（本文中为 FastAPI）负责执行任意需要的逻辑，为代码提供这些依赖（「注入」依赖项）。
依赖注入常用于以下场景：

• 共享业务逻辑（复用相同的代码逻辑）
• 共享数据库连接
• 实现安全、验证、角色权限
  上述场景均可以使用依赖注入，将代码重复最小化。

创建依赖项

首先，要关注的是依赖项

#   创建依赖项
#   依赖项就是一个函数，且可以使用与路径操作函数相同的参数：
from typing import Optional
from fastapi import FastAPI,Depends     # 导入 Depends
app = FastAPI()

async def common_parameters(q:Optional[str] = None,skip:int=0,limit:int=100):
    return "q":q,"skip":skip,"limit":limit

@app.get("/items/")
async def read_items(common:dict = Depends(common_parameters())):           #   声明依赖项
    return common

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):    #声明依赖项；与在路径操作函数参数中使用 Body、Query 的方式相同，声明依赖项需要使用 Depends 和一个新的参数：
    return commons
#   依赖项函数的形式和结构与路径操作函数一样。
#   因此，可以把依赖项当作没有「装饰器」（即，没有 @app.get("/some-path") ）的路径操作函数。
#   依赖项可以返回各种内容。
#   本例中的依赖项预期接收如下参数：
#   类型为 str 的可选查询参数 q
#   类型为 int 的可选查询参数 skip，默认值是 0
#   类型为 int 的可选查询参数 limit，默认值是 100
#   然后，依赖项函数返回包含这些值的 dict。
#
#
#   接收到新的请求时，FastAPI 执行如下操作：
#   1.用正确的参数调用依赖项函数（「可依赖项」）
#   2.获取函数返回的结果
#   3.把函数返回的结果赋值给路径操作函数的参数
#·  这样，只编写一次代码，FastAPI 就可以为多个路径操作共享这段代码 。

简单用法

观察一下就会发现，只要路径 和操作匹配，就可以使用声明的路径操作函数。然后，FastAPI 会用正确的参数调用函数，并提取请求中的数据。实际上，所有（或大多数）网络框架的工作方式都是这样的。开发人员永远都不需要直接调用这些函数，这些函数是由框架（在此为 FastAPI ）调用的。通过依赖注入系统，只要告诉 FastAPI 路径操作函数 还要「依赖」其他在路径操作函数之前执行的内容，FastAPI 就会执行函数代码，并「注入」函数返回的结果。其他与「依赖注入」概念相同的术语为：

1. 资源（Resource）
2. 提供方（Provider）
3. 服务（Service）
4. 可注入（Injectable）
5. 组件（Component）

FastAPI 插件

依赖注入系统支持构建集成和「插件」。但实际上，FastAPI 根本不需要创建「插件」，因为使用依赖项可以声明不限数量的、可用于路径操作函数的集成与交互。

创建依赖项非常简单、直观，并且还支持导入 Python 包。毫不夸张地说，只要几行代码就可以把需要的 Python 包与 API 函数集成在一起。

FastAPI 兼容性

依赖注入系统如此简洁的特性，让 FastAPI 可以与下列系统兼容：

• 关系型数据库
• NoSQL 数据库
• 外部支持库
• 外部 API 认证和鉴权系统
• API 使用监控系统
• 响应数据注入系统

虽然，层级式依赖注入系统的定义与使用十分简单，但它却非常强大。
比如，可以定义依赖其他依赖项的依赖项。
最后，依赖项层级树构建后，依赖注入系统会处理所有依赖项及其子依赖项，并为每一步操作提供（注入）结果。
比如，下面有 4 个 API 路径操作（端点）：
/items/public/
/items/private/
/users/user_id/activate
/items/pro/
开发人员可以使用依赖项及其子依赖项为这些路径操作添加不同的权限：

类作为依赖项

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



class Cat:
    def __init__(self, name: str):
        self.name = name
fluffy = Cat(name="Mr Fluffy")
#   在这种情况下，fluffy是类的一个实例Cat;    而要创造fluffy，你就是在“召唤” Cat。因此，Python 类也是一个可调用的.
#
#   可以将依赖“可靠”common_parameters从上面更改为类CommonQueryParams：
from fastapi import Depends, FastAPI
app = FastAPI()

fake_items_db = ["item_name": "Foo", "item_name": "Bar", "item_name": "Baz"]

class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):      #   注意__init__用于创建类实例的方法：它与我们之前的参数相同common_parameters：
        self.q = q
        self.skip = skip
        self.limit = limit
#   这些参数是FastAPI用来“解决”依赖关系的参数。
#   在这两种情况下，它将具有：
#   一个可选的q查询参数，它是一个str.
#   skip查询参数为，int默认值为0。
#   limit查询参数为，int默认值为100。
#   在这两种情况下，数据都将被转换、验证、记录在 OpenAPI 模式中，等等。

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):      #   可以使用这个类来声明你的依赖了。
    response = 
    if commons.q:
        response.update("q": commons.q)
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update("items": items)
    return response
#   FastAPI调用CommonQueryParams该类。这将创建该类的“实例”，并且该实例将作为参数传递commons给您的函数。
#   类型注释与Depends
#   注意我们如何CommonQueryParams在上面的代码中写了两次：
#   commons: CommonQueryParams = Depends(CommonQueryParams)


#   实际上可以只写：
#
#
# commons = Depends(CommonQueryParams)
@app.get("/items/")
async def read_items(commons=Depends(CommonQueryParams)):
    response = 
    if commons.q:
        response.update("q": commons.q)
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update("items": items)
    return response


#   将依赖项声明为参数的类型，并将Depends()其用作该函数参数的“默认”值（在 之后），在=中没有任何参数Depends()，而不必在Depends(CommonQueryParams).
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends()):
    response = 
    if commons.q:
        response.update("q": commons.q)
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update("items": items)
    return response

子依赖项

#   子依赖项
#   下列代码创建了第一层依赖项：
from typing import Optional
from fastapi import Cookie, Depends, FastAPI
app = FastAPI()

def query_extractor(q: Optional[str] = None):
    return q
#   这段代码声明了类型为 str 的可选查询参数 q，然后返回这个查询参数。
# 这个函数很简单（不过也没什么用），但却有助于让我们专注于了解子依赖项的工作方式。

def query_or_cookie_extractor(          #  创建另一个依赖项函数，并同时用该依赖项自身再声明一个依赖项（所以这也是一个「依赖项」）：
    q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(None)
):
    if not q:
        return last_query
    return q
#   这里重点说明一下声明的参数：
# 尽管该函数自身是依赖项，但还声明了另一个依赖项（它「依赖」于其他对象）
# 该函数依赖 query_extractor, 并把 query_extractor 的返回值赋给参数 q
# 同时，该函数还声明了类型是 str 的可选 cookie（last_query）
# 用户未提供查询参数 q 时，则使用上次使用后保存在 cookie 中的查询

@app.get("/items/")
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 返回的结果。
#   query_extractor -> query_or_cookie_extractor -> /items/



#   多次使用同一个依赖项
#   如果在同一个路径操作 多次声明了同一个依赖项，例如，多个依赖项共用一个子依赖项，FastAPI 在处理同一请求时，只调用一次该子依赖项。
#
# FastAPI 不会为同一个请求多次调用同一个依赖项，而是把依赖项的返回值进行「缓存」，并把它传递给同一请求中所有需要使用该返回值的「依赖项」。
#
# 在高级使用场景中，如果不想使用「缓存」值，而是为需要在同一请求的每一步操作（多次）中都实际调用依赖项，可以把 Depends 的参数 use_cache 的值设置为 False :
# async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):
#     return "fresh_value": fresh_value

路径操作装饰器依赖项

有时，我们并不需要在路径操作函数中使用依赖项的返回值。
或者说，有些依赖项不返回值。
但仍要执行或解析该依赖项。
对于这种情况，不必在声明路径操作函数的参数时使用 Depends，而是可以在路径操作装饰器中添加一个由 dependencies 组成的 list

#   在路径操作装饰器中添加 dependencies 参数
#   路径操作装饰器支持可选参数 ~ 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 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("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return ["item": "Foo", "item": "Bar"]
#   路径操作装饰器依赖项（以下简称为“路径装饰器依赖项”）的执行或解析方式和普通依赖项一样，但就算这些依赖项会返回值，它们的值也不会传递给路径操作函数。
#   有些编辑器会检查代码中没使用过的函数参数，并显示错误提示。
# 在路径操作装饰器中使用 dependencies 参数，可以确保在执行依赖项的同时，避免编辑器显示错误提示。
# 使用路径装饰器依赖项还可以避免开发新人误会代码中包含无用的未使用参数。
#
#
#
#   依赖项错误和返回值
#   路径装饰器依赖项也可以使用普通的依赖项函数。
#   依赖项的需求项
#   路径装饰器依赖项可以声明请求的需求项（比如响应头）或其他子依赖项：
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 HTTPException(status_code=400, detail="X-Token header invalid")           #   路径装饰器依赖项与正常的依赖项一样，可以 raise 异常.

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("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return ["item": "Foo", "item": "Bar"]

全局依赖项

有时，我们要为整个应用添加依赖项。
通过与定义路径装饰器依赖项 类似的方式，可以把依赖项添加至整个 FastAPI 应用。
这样一来，就可以为所有路径操作应用该依赖项：

#   全局依赖项
#   有时，我们要为整个应用添加依赖项
#   通过与定义路径装饰器依赖项 类似的方式，可以把依赖项添加至整个 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

app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])        #  通过与定义路径装饰器依赖项 类似的方式，可以把依赖项添加至整个 FastAPI 应用

@app.get("/items/")
async def read_items():
    return ["item": "Portal Gun", "item": "Plumbus"]

@app.get("/users/")
async def read_users():
    return ["username": "Rick", "username": "Morty"]

有产量的依赖

#   有产量的依赖
#   FastAPI 支持在完成后执行一些额外步骤的依赖项。
#   为此，请使用yield代替return，并在之后编写额外的步骤。
#   任何有效的函数都可以用于：
#   @contextlib.contextmanager
#   @contextlib.asynccontextmanager
#   作为FastAPI依赖项使用是有效的。
#   事实上，FastAPI 在内部使用了这两个装饰器。
#
#
#   数据库依赖项yield
#   例如，您可以使用它来创建数据库会话并在完成后关闭它。
#   yield在发送响应之前，仅执行语句之前和包含语句的代码：
async def get_db():
    db = DBSession()    #   yield在发送响应之前，仅执行语句之前和包含语句的代码：
    try:
        yield db        #   产生的值是注入到路径操作和其他依赖项中的值：
    finally:            #   yield响应发送后执行语句后面的代码：
        db.close()
#   依赖于yieldandtry
#   如果您try在依赖项中使用块yield，您将收到使用依赖项时引发的任何异常。
#
# 例如，如果某些代码在中间某个点、另一个依赖项或路径操作中，使数据库事务“回滚”或创建任何其他错误，您将在依赖项中收到异常。
#   因此，您可以使用except SomeException.
#
# 同样，您可以使用finally它来确保执行退出步骤，无论是否有异常。
  async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()


#   子依赖项yield
#   您可以拥有任何大小和形状的子依赖项和子依赖项的“树”，并且它们中的任何一个或全部都可以使用yield.
#   FastAPI将确保每个依赖项中的“退出代码”以yield正确的顺序运行。
#   例如，dependency_c可以依赖,dependency_b和:dependency_bdependency_a
from fastapi import Depends
async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a=Depends(dependency_a)):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b=Depends(dependency_b)):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)
#   并且他们都可以使用yield.
#
# 在这种情况下dependency_c，要执行其退出代码，需要来自dependency_b（此处命名为dep_b）的值仍然可用。
#
# 并且，反过来，dependency_b需要来自dependency_a（此处命名为dep_a）的值可用于其退出代码。
from fastapi import Depends


async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a=Depends(dependency_a)):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b=Depends(dependency_b)):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)
#   同样，您可以依赖yield和return混合。
#
# 并且您可能有一个依赖项，该依赖项需要多个其他依赖项yield，等等。
#
# 您可以拥有所需的任何依赖项组合。
#
# FastAPI将确保一切都以正确的顺序运行。



#   可以在FastAPI依赖项中使用它们，方法是在依赖项函数中yield使用 withorasync with语句：
class MySuperContextManager:
    def __init__(self):
        self.db = DBSession()

    def __enter__(self):
        return self.db

    def __exit__(self, exc_type, exc_value, traceback):
        self.db.close()


async def get_db():
    with MySuperContextManager() as db:
        yield db