python 大型项目神器实战

2022-11-21 12:31:56

在 fastapi 中有一个强大并且简单易懂的系统：依赖注入系统

1.什么是依赖注入

依赖注入首先意味着在程序中我们的代码可以声明一些它必须依赖的项：我们称之为 dependencies，也就是依赖项。然后，在实际运行中，fastapi 会把所有需要的依赖项提供给你的代码，称之为注入依赖项。

一句话简单解释:原本接受各种参数来构造一个对象，现在只接受一个参数这个参数是已经实例化的对象，对象的『依赖』是注入进来的，而和它的构造方式解耦了。构造它这个『控制』操作也交给了第三方。

依赖注入适用场景列举：

1.业务逻辑复用
2.共享数据库连接
3.安全机制、权限校验、角色管理等

所有上述使用场景，借助于依赖注入可

2.依赖注入实现方案

在看具体的实现依赖注入之前我们先从流程上来理解一下整个数据流转。当一个新的请求到来的时候，实际的调用流程如下:

1.调用依赖项函数(传递合适的参数)

2.得到依赖项目函数的返回结果

3.把返回结果传递给路由函数中对应的参数

4.路由函数中业务流数据处理

5.获取的数据返回给客户端/前端

函数级依赖项

# -*- encoding: utf-8 -*-

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_paras(request_source: str = None, page: int = 0, limit: int = 100):
    return {"request_source": request_source, "page": page, "limit": limit}


@app.get("/items/info")
async def read_items_info(commons: dict = Depends(common_paras)):
    return commons


@app.get("/users/info")
async def read_users_info(commons: dict = Depends(common_paras)):
    return commons

类级依赖项

# -*- encoding: utf-8 -*-
from fastapi import Depends, FastAPI

app = FastAPI()


fake_items_db = [{"user_name": "HaiShiNiu"}, {"user_name": "WangXiaoXiao"}, {"user_name": "ZhongCen"}]


class CommonQueryParams:
    def __init__(self, request_source: str = None, page: int = 0, limit: int = 100):
        self.request_source = request_source
        self.page = page
        self.limit = limit


@app.get("/items/info")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    response = {}
    if commons.request_source:
        response.update({"q": commons.q})
    items = fake_items_db[commons.page : commons.page + commons.limit]
    response.update({"items": items})
    return response

递归依赖项

# -*- encoding: utf-8 -*-
from fastapi import Cookie, Depends, FastAPI

app = FastAPI()


def query_extractor(q11: str, q12: str):
    return q11 + q12


def query_or_cookie_extractor(q2: str = Depends(query_extractor), last_query: str = Cookie(None)):
    if not q2:
        return last_query
    return q2


@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
return {"q_or_cookie": query_or_default}

注意:有些场景是需要对依赖项的多次调用的情况。

如果某个依赖项在同一个路径操作中被声明了多次，例如，多个依赖项都有一个共同的子依赖项，那么 fastapi 默认在每一次请求中只会调用这个依赖项一次。fastapi 会把这个依赖项的返回值缓存起来，然后把这个值传递给需要的依赖项，而不是在同一个请求中多次调用这个依赖项。在有些场景下，我们并不需要缓存这个依赖项的返回值，而是需要多次调用，那么我们可以使用参数 use_cache=False 来禁止依赖项的缓存。

async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):
    return {"fresh_value": fresh_value}

基于路径操作装饰器依赖项

在有些情况下，我们并不需要依赖项的返回值，但仍然需要依赖项被执行。在这种情况下，我们可以通过路径操作装饰器来操作依赖项的一个列表。路径操作装饰器接收一个可选的参数 dependencies，参数内容是 Depends()列表。

# -*- encoding: utf-8 -*-
from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


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


async def verify_key(x_key: str = Header(...)):
    if x_key != "key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


@app.get("/items/info", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items_info():

    return [{"function_name": "Foo"}, {"function_name": "Bar"}]

这些依赖项与普通依赖项的执行相同，但他们的返回值(如果有)不会被传递给路径操作函数。我们可以重复使用已经声明的依赖项，无论他们是否有返回值，都不会影响依赖项的执行。

带有 yield 功能依赖项

fastapi 支持依赖项在请求结束后做一些额外的工作。要实现这个功能，我们需要用 yield 代替 return，然后其后添加一些额外的工作。我们举个例子:创建一个数据库链接，然后在请求结束后关闭这个链接。当然处理这个 case 的解决方案也是有很多的，本次我们就基于使用 yield 功能的依赖项进行实现

# -*- encoding: utf-8 -*-

async def get_db_info():
    """
    获取数据库连接信息
    """
    # 创建链接
    db = DBSession()
    try:
        yield db  # 数据库使用
    finally:
        db.close()  # 关闭数据库链接

简单说明一下:yield db 后面的值 db 会注入给路径操作或者其他依赖项。yield db 后面的代码在 response 提交之后才会执行。使用 try 语句来捕获可能发生的异常。为了确保无论是否有异常发生都能执行退出逻辑，我们这里在 finally 语句中执行退出逻辑。但是要注意，如果尝试在 yield 后面抛出 HTTPException，不会起到任何作用。yield 之后的退出代码是在异常处理器之后被执行的，因此无法捕捉异常的发生。

可参数化依赖项

我们前面使用的依赖项都是固定的函数或者类，但有时想在依赖项中设置不同的参数，同时又不用声明不同的函数或类。此时可利用一个可调用的类实例来实现这个功能。注意:类本身就是可调用的，而它的实例需要实现一个特定类方法才是可调用的：call

一句话介绍 call：call 是 Python 的一个黑魔法方法，核心功能是通过对象实例可以直接触发 call 中的逻辑，在 Tornado 源码中这种用法很多，有兴趣的同学可以看看我在公众号中的关于 Tornado 的专栏文章。

# -*- encoding: utf-8 -*-

from fastapi import Depends, FastAPI

app = FastAPI()


class FixedContentQueryCheck(object):
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryCheck("bar")


@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}

总结

1.简单介绍了一下什么是依赖注入以及使用场景

2.我们系统的梳理了实现依赖注入的各种方式供大家在生产环境中进行使用

码农公寓