aiodine

aiodine为Python 3.6+提供了类似依赖注入的Pytest fixtures风格的异步优先依赖注入。

• 安装
• 概念
• 用法
• 常见问题解答
• 变更日志

安装

pip install "aiodine==1.*"

概念

aiodine围绕两个概念展开

• 提供者 负责设置、返回和可选地清理 资源。
• 消费者 可以通过声明提供者为它们的参数之一来访问这些资源。

这种方法是 依赖注入 的实现，使提供者和消费者

• 明确：在消费者签名上通过名称引用提供者使依赖关系清晰且可预测。
• 模块化：提供者本身也可以消费其他提供者，允许构建可重用（并可替换）的依赖关系生态系统。
• 灵活：提供值在给定的作用域内重用，提供者和消费者支持各种语法（异步/同步、函数/生成器），使供应变得有趣。

aiodine在以下意义上是 异步优先

• 它被设计为与协程函数和 async/await 语法一起工作。
• 消费者只能在异步环境中调用。
• 但是，提供者和消费者函数也可以是常规Python函数和生成器，仅为了方便。

用法

提供者

供应商 在特定 范围 内将 资源 可用于消费者。它们通过使用 @aiodine.provider 装饰 供应商函数 来创建。

这里是一个 "Hello World" 供应商

import aiodine

@aiodine.provider
async def hello():
    return "Hello, aiodine!"

供应商可以在两个 范围 中使用

• 函数：供应商的值每次被消费时都会重新计算。
• 会话：供应商的值仅计算一次（第一次消费时），在后续调用中重用。

默认情况下，供应商是函数范围的。

消费者

一旦声明了供应商，就可以由 消费者 使用。通过使用 @aiodine.consumer 装饰 消费者函数 来构建消费者。消费者可以将供应商作为其参数之一声明，aiodine将在运行时注入它。

这里是一个消费者示例

@aiodine.consumer
async def show_friendly_message(hello):
    print(hello)

所有的 aiodine 消费者都是异步的，所以您需要在一个异步上下文中运行它们

from asyncio import run

async def main():
    await show_friendly_message()

run(main())  # "Hello, aiodine!"

当然，消费者也可以声明非供应商参数。aiodine 足够智能，可以确定哪些参数应通过供应商注入，哪些应从调用者期望。

@aiodine.consumer
async def show_friendly_message(hello, repeat=1):
    for _ in range(repeat):
        print(hello)

async def main():
    await show_friendly_message(repeat=10)

消费其他供应商的供应商

供应商是模块化的，因为它们可以自己消费其他供应商。

然而，为了使其工作，供应商需要首先 冻结。这确保了无论声明顺序如何，依赖图都能正确解析。

import aiodine

@aiodine.provider
def email():
    return "user@example.net"

@aiodine.provider
async def send_email(email):
    print(f"Sending email to {email}…")

aiodine.freeze()  # <- Ensures that `send_email` has resolved `email`.

注意：多次调用 .freeze() 是安全的。

还提供了上下文管理器语法

import aiodine

with aiodine.exit_freeze():
    @aiodine.provider
    def email():
        return "user@example.net"

    @aiodine.provider
    async def send_email(email):
        print(f"Sending email to {email}…")

生成器供应商

生成器供应商可以用于在供应商超出范围后执行清理（最终化）操作。

import os
import aiodine

@aiodine.provider
async def complex_resource():
    print("setting up complex resource…")
    yield "complex"
    print("cleaning up complex resource…")

提示：即使消费者中发生异常，清理代码也会执行，因此不需要在 yield 语句周围使用 try/finally 块。

重要：会话范围的生成器供应商仅在会话上下文中使用时才会进行清理。有关详细信息，请参阅 会话。

延迟异步供应商

异步供应商默认是 急切 的：它们的返回值在注入到消费者之前会被等待。

您可以通过标记供应商为 延迟 来将提供的值的等待延迟到消费者。当供应商需要条件评估时，这很有用。

from asyncio import sleep
import aiodine

@aiodine.provider(lazy=True)
async def expensive_io_call():
    await sleep(10)
    return 42

@aiodine.consumer
async def compute(expensive_io_call, cache=None):
    if cache:
        return cache
    return await expensive_io_call

工厂供应商

工厂供应商返回一个 函数，而不是返回标量值。工厂供应商对于实现接受各种输入的可重用供应商非常有用。

这更多是一个 设计模式。实际上，aiodine 没有额外的代码来支持这个功能。

以下示例定义了一个用于（模拟）数据库查询的工厂供应商

import aiodine

@aiodine.provider(scope="session")
async def notes():
    # Some hard-coded sticky notes.
    return [
        {"id": 1, "text": "Groceries"},
        {"id": 2, "text": "Make potatoe smash"},
    ]

@aiodine.provider
async def get_note(notes):
    async def _get_note(pk: int) -> list:
        try:
            # TODO: fetch from a database instead?
            return next(note for note in notes if note["id"] == pk)
        except StopIteration:
            raise ValueError(f"Note with ID {pk} does not exist.")

    return _get_note

在消费者中的示例用法

@aiodine.consumer
async def show_note(pk: int, get_note):
    print(await get_note(pk))

提示：您可以将工厂供应商与 生成器供应商 结合使用，以清理工厂需要使用的任何资源。以下是一个提供临时文件并在清理时删除它们的示例

import os
import aiodine

@aiodine.provider(scope="session")
def tmpfile():
    files = set()

    async def _create_tmpfile(path: str):
        with open(path, "w") as tmp:
            files.add(path)
            return tmp

    yield _create_tmpfile

    for path in files:
        os.remove(path)

在不声明为参数的情况下使用供应商

有时，消费者需要使用供应商，但不关心它返回的值。在这些情况下，您可以使用 @useprovider 装饰器并跳过将其声明为参数。

提示：@useprovider 装饰器接受可变数量的供应商，可以按名称或按引用提供。

import os
import aiodine

@aiodine.provider
def cache():
    os.makedirs("cache", exist_ok=True)

@aiodine.provider
def debug_log_file():
    with open("debug.log", "w"):
        pass
    yield
    os.remove("debug.log")

@aiodine.consumer
@aiodine.useprovider("cache", debug_log_file)
async def build_index():
    ...

自动使用供应商

自动使用供应商会在其配置的范围 自动激活（无需在消费者中声明为参数）。

这通常可以节省您装饰所有消费者以使用 @useprovider。

例如，以下自动使用的提供者会在每次调用消费者时将当前日期和时间打印到控制台。

import datetime
import aiodine

@aiodine.provider(autouse=True)
async def logdatetime():
    print(datetime.now())

会话

会话是会话提供者存在的上下文。

更具体地说，会话提供者（或生成器会话提供者）在进入会话时实例化（或设置），在退出会话时销毁（或清理）。

要进入会话，请使用

await aiodine.enter_session()

要退出它

await aiodine.exit_session()

异步上下文管理器语法也是可用的

async with aiodine.session():
    ...

上下文提供者

警告：这是一个实验性功能。

上下文提供者被引入以解决注入上下文局部资源的问题。这些资源在提供者声明时通常是未定义的，但在进入某种类型的上下文时变得明确。

这听起来可能很抽象，所以在展示上下文提供者的用法之前，让我们先看看一个例子。

示例

假设我们在一家餐厅。在那里，服务员执行顾客提交的订单。每个顾客都会得到一个Order对象，他们可以将他们想要的菜单项写入。

在aiodine术语中，服务员是提供者，而顾客是消费者。

在服务期间，服务员需要监听新顾客，创建一个新的Order对象，将其提供给顾客，执行顾客所写的订单，并销毁已执行的订单。

因此，在这个例子中，上下文从创建订单到销毁订单的范围，并且对于特定的顾客是特定的。

以下是模拟服务员这一方面的代码可能的样子

from asyncio import Queue

import aiodine

class Order:
    def write(self, item: str):
        ...

class Waiter:
    def __init__(self):
        self._order = None
        self.queue = Queue()

        # Create an `order` provider for customers to use.
        # NOTE: the actually provided value is not defined yet!
        @aiodine.provider
        def order():
            return self._order

    async def _execute(self, order: Order):
        ...

    async def _serve(self, customer):
        # NOTE: we've now entered the *context* of serving
        # a particular customer.

        # Create a new order that the customer can
        # via the `order` provider.
        self._order = Order()

        await customer()

        # Execute the order and destroy it.
        await self._execute(self._order)
        self._order = None

    async def start(self):
        while True:
            customer = await self.queue.get()
            await self._serve(customer)

需要注意的是，顾客可以对订单做任何事情。特别是，他们可能需要一些时间来考虑他们要订购的内容。同时，服务器将监听其他顾客的调用。从这个意义上说，这种情况是一个异步的。

一个可能的顾客代码可能如下所示

from asyncio import sleep

@aiodine.consumer
def alice(order: Order):
    # Pondering while looking at the menu…
    await sleep(10)
    order.write("Pizza Margheritta")

让我们稍微反思一下。你注意到服务员只持有一个对Order的引用吗？这意味着只要一次只为一个顾客服务，代码就可以正常工作。

但如果另一个顾客，比如bob，在alice考虑要订购什么的时候出现，根据当前的实现，服务员将简单地忘记alice的订单，并最终执行bob的订单两次。简而言之：我们将遇到一个竞争条件。

通过使用上下文提供者，我们可以透明地将服务员的对订单的引用转换为一个上下文变量（即ContextVar）。它是每个顾客上下文本地的，这解决了竞争条件。

以下是代码可能的样子

import aiodine

class Waiter:
    def __init__(self):
        self.queue = Queue()
        self.provider = aiodine.create_context_provider("order")

    async def _execute(self, order: Order):
        ...

    async def _serve(self, customer):
        order = Order()
        with self.provider.assign(order=order):
            await customer()
            await self._execute(order)

    async def start(self):
        while True:
            customer = await self.queue.get()
            await self._serve(customer)

注意

• 顾客可以使用order提供者，就像以前一样。实际上，它是在调用.create_context_provider()时创建的。
• order现在是上下文局部的，即如果其他顾客同时下单，其值不会被遗忘或混淆。

这种情况可能对一些人来说看起来很平常，但它很可能在客户端/服务器架构中找到，包括在Web框架中。

用法

要创建上下文提供者，请使用aiodine.create_context_provider()。此方法接受可变数量的参数，并返回一个ContextProvider。每个参数都用作新@provider的名称，该提供者提供ContextVar对象的内容。

import aiodine

provider = aiodine.create_context_provider("first_name", "last_name")

每个上下文变量最初都包含None。这意味着消费者将收到None——除非它们在.assign()块中调用。

with provider.assign(first_name="alice"):
    # Consumers called in this block will receive `"alice"`
    # if they consume the `first_name` provider.
    ...

常见问题解答

为什么是"aiodine"?

aiodine 包含 "aio"，类似于 asyncio，以及 "di"，类似于 依赖注入。最后两个字母的组合使得 aiodine 的发音类似于 碘，这种化学元素。

变更日志

参见 CHANGELOG.md。

许可

MIT