【python】FastAPI依赖注入（四）

7-1 什么是依赖注入

场景：

你的一个网站有两个查询接口，一个是图书列表借口，一个是用户列表借口，两个接口有相同的分页查询逻辑，此时你该如何实现？看看下面的普通代码实现方式

from fastapi import FastAPI

app = FastAPI()


BOOKS = [{"id": i, "name": f"book{i}", "status": i % 4 != 0} for i in range(1, 11)]
USERS = [{"id": i, "name": f"user{i}", "status": i % 4 != 0} for i in range(1, 11)]


@app.get("/api/books")
def get_books(page: int = 1, size: int = 2, status: bool = True):  # 需要定义三个参数 page\size\status
    books = [b for b in BOOKS if b["status"] == status]
    return books[(page - 1) * size:page * size]


@app.get("/api/users")
def get_users(page: int = 1, size: int = 2, status: bool = True):  # 需要重复再次定义三个参数 代码重复写了
    users = [u for u in USERS if u["status"] == status]
    retur

对于上面的代码重复问题，我们可以使用依赖注入解决
依赖注入其实就是英文单词 Dependency Injection 的翻译，它是一种非常简单且直观的工具。简单来说，就是当你的代码中的一个参数的值需要依赖其他条件时，可以使用依赖注入。当你使用了它之后，FastAPI就会在需要使用这个参数的时候，自动帮你执行它的依赖条件来获取结果。
- 函数的入参需要符合某一种规范（所谓依赖）。

依赖注入的使用

首先，从FastAPI引入Depends

然后，定义依赖条件，即定义一个函数（可调用对象即可），函数的形参是我们需要的参数，返回值是三个参数组成的字典

最后，在路径函数内使用 Depends(common_params), 这样一来FastAPI就知道，commons这个字典依赖common_params函数的返回值，即commons字典就是common_params函数的返回值。

像common_params函数被称为依赖条件，只要是可调用对象就可以当依赖条件

依赖条件中可以使用Path\Query等等获取参数的方式。

from fastapi import FastAPI, Depends

app = FastAPI()


BOOKS = [{"id": i, "name": f"book{i}", "status": i % 4 != 0} for i in range(1, 11)]
USERS = [{"id": i, "name": f"user{i}", "status": i % 4 != 0} for i in range(1, 11)]


# 定义依赖条件
def common_params(page: int = 1, size: int = 2, status: bool = True):
    return {
        "page": page,
        "size": size,
        "status": status,
    }

# 下面两个路径函数都是使用了上面定义的依赖条件，依赖条件函数的入参就是洗面路径函数的入参要求，且默认值就是其默认值
@app.get("/api/books")
def get_books(commons: dict = Depends(common_params)):	# 使用依赖条件
    page = commons["page"]
    size = commons["size"]
    status = commons["status"]
    books = [b for b in BOOKS if b["status"] == status]
    return books[(page - 1) * size:page * size]


@app.get("/api/users")
def get_users(commons: dict = Depends(common_params)):
    page = commons["page"]
    size = commons["size"]
    status = commons["status"]
    users = [u for u in USERS if u["status"] == status]
    return users[(page - 1) * size:page * size]

依赖注入的用途：

共享一块相同逻辑的代码块
共享数据库连接
权限认证，登录状态认证
等等等

7-2 依赖注入嵌套使用【】

依赖注入是非常强大的，比如说，它可以支持嵌套使用，且嵌套深度不受限制

示例：两层嵌套依赖注入

路径函数get_name需要的形参username_or_nickname有依赖条件，所以FastAPI会调用 username_or_nickname_extractor
执行username_or_nickname_extractor的时候，发现它也有依赖条件，所以FastAPI会调用 username_extractor
按照这个顺序，依次获取每个有依赖条件的参数的结果。最终，在路径函数内获取最终的结果。

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()
# 这里的Union[str, None] 函数的意思是指定多个类型中的一个，也就是说指定str类型或者None类型 Union是联合的意思

# username_extractor函数的入参是一个，
def username_extractor(username: Union[str, None] = None):
    return username

# username_or_nickname_extractor这个函数的入参有两个，第一个依赖了username_extractor这个函数的入参，第二个是nickname
def username_or_nickname_extractor(
    username: str = Depends(username_extractor),
    nickname: Union[str, None] = None,
):
    if not username:
        return nickname
    return username



# 这里路径函数的入参是依赖了username_or_nickname_extractor这个函数的入参，
@app.get("/name")
def get_name(username_or_nickname: str = Depends(username_or_nickname_extractor)): 
    return {"username_or_nickname": username_or_nickname}

7-3 依赖注入的缓存现象

很多时候，我们定义的依赖条件会被执行多次，这种场景下，FastAPI默认只会执行一次依赖条件。但我们也可以执行不使用缓存。

示例1：依赖注入的缓存现象

依赖条件get_num被依赖了两次，但是你会发现其内部打印语句只打印了一次。也就是说，第二次使用这个依赖条件时FastAPI并没有真正执行这个函数，而是直接使用了第一次执行的结果，这就是依赖注入的缓存现象。

from fastapi import Depends, FastAPI

app = FastAPI()


def get_num(num: int):
    print("get_num被执行了")
    return num


@app.get("/")
def get_results(num1: int = Depends(get_num), num2: int = Depends(get_num)):
    return {"num1": num1, "num2": num2}

实例2：依赖注入不使用缓存

默认 use_cache字段是True，如果在第二次使用依赖注入不想使用缓存，将此字段的值设为False即可。
需要注意，

1
2
3

@app.get("/")
def get_results(num1: int = Depends(get_num), num2: int = Depends(get_num, use_cache=False)):
    return {"num1": num1, "num2": num2}

示例3：缓存现象存在缓存嵌套中

依赖注入嵌套使用时，子依赖如果被使用多次也会存在缓存现象，解决办法就是第二次使用子依赖时使用use_cache=False

from fastapi import Depends, FastAPI

app = FastAPI()

# 这个子依赖被使用两次
def get_num(num: int):
    print("get_num被执行了")
    return num


def get_result1(num: int = Depends(get_num)):
    return num * num


# 第二次使用get_num时，设置use_cache=False则不使用缓存
def get_result2(num: int = Depends(get_num, use_cache=False)):
    return num * num * num


@app.get("/")
def get_results(result1: int = Depends(get_result1), result2: int = Depends(get_result2)):
    return {"result1": result1, "result2": result2}

总结：

在一个请求中，如果依赖注入条件被使用了多次，则只有第一次会真正执行内部代码，然后将其返回值缓存起来，后面再次使用它则直接获取缓存结果，不会再次执行其内部代码。
如果不想使用依赖注入缓存，则可以在这个依赖条件第二次被使用时，设置 use_cache=False即可。

7-4 路径装饰器和全局依赖注入【有用】

有的时候，我们想使用依赖注入，但并不希望它有返回值，那此时就不能在路径函数内使用Depends()了，那该如何呢？

不需要返回值的依赖注入，可以直接在路径装饰器中使用

FastAPI的解决方式

不需要返回值的依赖注入，可以在路径装饰器中使用。

示例1：使用依赖注入校验访问权限

在路径装饰器中可以使用依赖注入，使用字段dependencies，它的值需要是一个包含Depands()的序列，比如列表或元组。
dependencies中每个依赖条件不需要返回值，就是有返回值也不会使用。

from fastapi import Depends, FastAPI, Header, HTTPException


app = FastAPI()


def verify_token(x_token: str = Header()):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")
    return x_token


@app.get("/items/", dependencies=[Depends(verify_token)]) # 这个时候，在接口文档上就可以看到这条接口是需要 x_token 入参的，但是不需要返回值
# x_token的来源是依赖注入接口的结果verify_token
def get_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

示例2：全局使用依赖注入校验访问权限

如果我们希望系统中的所有路由接口都默认有依赖注入，此时我们没有必要每个接口都重复设置一遍依赖注入
只需要在实例化FastAPI时，通过dependencies,需要注意的是，这个参数接收的值依然是一个包含多个可依赖对象的列表。

from fastapi import Depends, FastAPI, Header, HTTPException


def verify_token(x_token: str = Header()):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


app = FastAPI(dependencies=[Depends(verify_token)])
# 在实例化FastAPI的时候选择依赖注入，这样这个实例化对象的路径装饰器装饰的路径函数都会需要注入依赖

@app.get("/items/")
def get_items():
    return [{"item": "Portal Gun"}, {"item": "Plumbus"}]


@app.get("/users/")
def get_users():
    return [{"username": "Rick"}, {"username": "Morty"}]

7-5 基于类的依赖注入【有点用】

在7-1节中，我们通过定义依赖条件，解决了通用查询参数重复定义的问题。其中，我们定义的依赖条件是普通的函数。

但其实，只要是可调用的对象，都可以当做依赖条件，比如类。类实例化其实就是类的调用。

示例1：类形式的依赖条件

CommonQueryParams()就会实例化出来一个对象，三个通用参数就会保存在在对象的三个属性上。
类中定义的__init__方法，用来保存变量属性。

from fastapi import Depends, FastAPI

app = FastAPI()


BOOKS = [{"id": i, "name": f"book{i}", "status": i % 4 != 0} for i in range(1, 11)]


# 定义依赖条件
class CommonQueryParams:
    def __init__(self, page: int, size: int, status: bool):
        self.page = page
        self.size = size
        self.status = status


@app.get("/api/books")
def get_books(commons: CommonQueryParams = Depends(CommonQueryParams)):	    # 使用依赖条件 
# 这里 CommonQueryParams类的init属性就是入参依赖，

    page = commons.page
    size = commons.size
    books = [b for b in BOOKS if b["status"] == commons.status]
    return books[(page - 1) * size:page * size]

示例2：类形式依赖注入的简化用法

上面我们使用类形式的依赖注入，看起来优点冗余对吧：commons: CommonQueryParams = Depends(CommonQueryParams)
对于这种情况，有一个简写方式：commons: CommonQueryParams = Depends()，此时fastapi知道Depends括号内依赖的是什么。

# 定义依赖条件
class CommonQueryParams:
    def __init__(self, page: int, size: int, status: bool):
        self.page = page
        self.size = size
        self.status = status


@app.get("/api/books")
def get_books(commons: CommonQueryParams = Depends()): #简单一点写就是直接类的实例化对象，CommonQueryParams = Depends()就是在实例化这个类
    pass

7-6 基于对象的依赖注入【常用】

上面我们可以定义基于函数，或者基于类形式的依赖条件，但是他们有一个共同的缺点：写死不可变的，不支持参数化。

比如，我们有一个需求，要求我们检查查询参数中的字段是否包含指定的文本，并且被检查的文本可以通过参数的形式调整。

示例1：检查指定的文本是否在查询参数q中

缺点：代码重复，不够灵活优雅

from fastapi import  FastAPI

app = FastAPI()


@app.get("/hello")
def hello_check(q: str):
    exists = "hello" in q
    return {"exists": exists}


@app.get("/world")
def world_check(q: str): # 入参是q 判断q包不包含world ，包含就返回ture,否这flase
    exists = "world" in q
    return {"exists": exists}

FastAPI的解决方式

对于上面的的需求，我们可以使用基于对象的依赖注入，只要在类下面实现__call__方法，该对象就可以被调用，基于可以当做依赖条件

示例2：基于对象的依赖条件，参数化

对象FixedContentQueryChecker("hello")当依赖条件，它被调用时就会执行其魔法方法__call__，就会判断查询参数q中是否包含待检查的文本信息。

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):     # 实例化对象是执行
        self.fixed_content = fixed_content

    def __call__(self, q: str = "") -> bool:    # 对象被调用时执行
        return self.fixed_content in q

# FixedContentQueryChecker("hello")是在将类当方法调用，此时就会触发类的call方法
@app.get("/hello")
def hello_check(exists: bool = Depends(FixedContentQueryChecker("hello"))):
    return {"exists": exists}


@app.get("/world")
def world_check(exists: bool = Depends(FixedContentQueryChecker("world"))):
    return {"exists": exists}

这里复习一下python魔法方法call

类中的call在对象被调用时出发。就是当对象加括号被调用时触发
对象或者变量，只有实现了call方法，才是可调用对象，才可以被执行，否则就会报错， object is not callable
定义了call后，你实例化的对象可以像函数一样被调用，

7-7 依赖注入使用yield【好玩】

fastapi的依赖注入非常强大，它还可以具备上下文管理器的功能。
想要在依赖注入中实现上下文管理器，我们可以使用 yield
比如，我们想要在进入路径操作函数时通过依赖获取一个操作数据库的连接，请求结束后关闭这个db连接

示例1：依赖注入中使用yield

这样写的好处是，在路径操作函数结束时，会自动关闭db连接回收资源。及时在路径函数会出现异常报错，最终也会关闭连接。

from fastapi import Depends, FastAPI


app = FastAPI()


def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()


@app.get("/books/")
def books(db: Depend(get_db)): # 和pytest的fixture类似使用，yield函数就是迭代器
    # db.execute("select * form books")   db操作
    pass

7-8 回忆补充上下文管理器【重要】

上下文管理器是什么？

是一种用于管理资源的机制
它提供了一种方便的方式来自动获取和释放资源，确保资源在使用完毕后被正确的释放，无论是正常执行还是发生异常，都能正确释放

上下文管理协议需要实现两个功能：

__enter__，进入上下文时运行，并返回当前对象。如果with语句有as关键词存在，返回值会绑定在as后的变量上。
__exit__，退出上下文时运行

上下文管理器中的 with, 它的一个常见使用场景如下：

open函数会返回一个文件类型变量，这个文件类实现了上下文管理协议，而with语句就是为支持上下文管理器而存在的。

1 2	`with open("test.txt","r") as f: # open()返回的就是一个上下文管理器 content = f.read()`

进入with语句块时，就会执行文件类的__enter__返回一个文件对象，并赋值给变量 f
从with语句块出来时，机会执行文件类的__exit__，在其内部实现 f.close()，所以使用者就不需要在手动关闭文件对象了。

示例1：手动封装实现类似于open()的上下文管理器

class MyFile():
    def __init__(self,name):
        self.name = name
    def __enter__(self):
        print("进入with......")
        return self   # 返回的数据复制给下面的p
    def __exit__(self， exc_type, exc_val, exc_tb): 
        # exc_type表示异常类型，exc_val表示异常值，exc_tb表示异常回调 是默认要填的
        print("退出with=====")
        
with MyFile('Tom') as p:  # MyFile（ ）返回的也是一个上下文管理器
    print(p.name)  
# 执行的结果是
# 进入with......
# Tom
# 退出with=====

示例2：contextlib 模块实现上下文管理器【拓展】

使用装饰器 contextmanager
get_file函数内部，yield语句前的代码在进入with语句时执行，yield的值赋值给 as后面的变量，
yield后面的代码在退出with语句时执行

import contextlib


@contextlib.contextmanager
def get_file(filename: str):
    file = open(filename, "r", encoding="utf-8")
    yield file
    file.close()


with get_file('test.txt') as f:
    print(f.read()) # 打印文件内的内容

python

#python #编程，web开发

【python】FastAPI依赖注入（四）

http://example.com/2024/01/18/684FastAPI4依赖注入/

作者

Wangxiaowang

发布于

2024年1月18日

许可协议

python 之 pytest 常用插件（八）上一篇

【python】FastAPI表单和文件上传（三）下一篇