asyncinject

使用pytest-fixtures风格的依赖注入运行异步工作流程

安装

使用pip安装此库

$ pip install asyncinject

用法

此库受pytest fixtures的启发。

想法是通过允许使用一组函数定义它们，其中函数参数表示需要首先执行的依赖函数来简化执行并行asyncio操作。

然后库可以创建并执行一个计划，以尽可能有效的顺序并行执行所需的函数。

以下是一个示例，使用httpx HTTP库。

from asyncinject import Registry
import httpx


async def get(url):
    async with httpx.AsyncClient() as client:
        return (await client.get(url)).text

async def example():
    return await get("http://www.example.com/")

async def simonwillison():
    return await get("https://simonwillison.net/search/?tag=empty")

async def both(example, simonwillison):
    return example + "\n\n" + simonwillison

registry = Registry(example, simonwillison, both)
combined = await registry.resolve(both)
print(combined)

如果您在ipython或python -m asyncio（在控制台中启用顶级await）中运行此代码，您将看到结合了两个页面HTML的输出。

将对www.example.com和simonwillison.net的HTTP请求并行执行。

库会注意到both()接受两个参数，这两个参数是其他注册的async def函数的名称，并将构建一个执行计划，该计划并行执行这两个函数，然后将它们的输出传递给both()方法。

Registry.from_dict()

将函数列表传递给Registry构造函数将为每个函数注册其内省的函数名，使用fn.__name__。

您可以使用字典设置显式的名称

registry = Registry.from_dict({
    "example": example,
    "simonwillison": simonwillison,
    "both": both
})

这些字符串名称将用于匹配参数，因此每个函数都需要接受以该字典中使用的键命名的参数。

注册其他函数

可以注册的函数可以是常规函数或async def函数。

除了通过构造函数传递函数来注册函数外，您还可以使用 .register() 方法将它们添加到注册表中

async def another():
    return "another"

registry.register(another)

要使用除函数名称以外的名称进行注册，请传递 name= 参数

async def another():
    return "another 2"

registry.register(another, name="another_2")

解决未注册的函数

您不需要注册传递给 .resolve() 的最终函数 - 如果传递未注册的函数，库将直接 introspect 函数的参数并解决它们。

这对常规和异步函数都有效

async def one():
    return 1

async def two():
    return 2

registry = Registry(one, two)

# async def works here too:
def three(one, two):
    return one + two

print(await registry.resolve(three))
# Prints 3

参数会传递下去

您的依赖函数可以要求传递给 .resolve() 调用的关键字参数

async def get_param_1(param1):
    return await get(param1)

async def get_param_2(param2):
    return await get(param2)

async def both(get_param_1, get_param_2):
    return get_param_1 + "\n\n" + get_param_2


combined = await Registry(get_param_1, get_param_2, both).resolve(
    both,
    param1 = "http://www.example.com/",
    param2 = "https://simonwillison.net/search/?tag=empty"
)
print(combined)

具有默认值的参数将被忽略

您可以通过为参数分配默认值来将其排除在依赖注入机制之外

async def go(calc1, x=5):
    return calc1 + x

async def calc1():
    return 5

print(await Registry(calc1, go).resolve(go))
# Prints 10

使用计时器进行跟踪

您可以将一个 timer= 可调用对象传递给 Registry 构造函数以收集关于执行任务的时间信息。您的函数应接受三个位置参数

• name - 正在被计时的函数的名称
• start - 它开始执行的时间，使用 time.perf_counter() (perf_counter() 文档)
• end - 它完成执行的时间

您也可以在这里使用 print

combined = await Registry(
    get_param_1, get_param_2, both, timer=print
).resolve(
    both,
    param1 = "http://www.example.com/",
    param2 = "https://simonwillison.net/search/?tag=empty"
)

这将输出

get_param_1 436633.584580685 436633.797921747
get_param_2 436633.641832699 436634.196364347
both 436634.196570217 436634.196575639

关闭并行执行

默认情况下，根据执行计划可以并行运行的函数将使用 asyncio.gather() 并行运行。

您可以通过将 parallel=False 传递给 Registry 构造函数或在创建注册表对象后设置 registry.parallel = False 来禁用此并行执行。

这主要用于比较您项目的并行和顺序执行之间的差异。

开发

要为此库做出贡献，首先检出代码。然后创建一个新的虚拟环境

cd asyncinject
python -m venv venv
source venv/bin/activate

现在安装依赖项并测试依赖项

pip install -e '.[test]'

要运行测试

pytest