quattro: asyncio任务控制

quattro是一个Apache 2许可证的库，用Python编写，用于asyncio应用程序中的任务控制。quattro受到来自Trio框架的结构化并发概念的启发。

quattro支持Python版本3.9 - 3.11，包括PyPy。

安装

要安装quattro，只需

$ pip install quattro

quattro.gather

quattro提供了一个基于任务组的独立、简单的asyncio.gather实现。与子任务中的错误相比，quattro版本更安全，并且在后台使用任务组以防止任务泄露。

from quattro import gather

async def my_handler():
    res_1, res_2 = await gather(long_query_1(), long_query_2())

可以使用return_exceptions参数来使用gather()捕获并返回异常作为响应，而不是让它们冒泡出去。

from quattro import gather

async def my_handler():
    res_1, res_2 = await gather(
        long_query_1(),
        long_query_2(),
        return_exceptions=True,
    )

    # res_1 and res_2 may be instances of exceptions.

quattro.gather()与asyncio.gather()的区别在于

• 如果一个子任务失败，其他未完成的任务将被取消，就像在TaskGroup中一样。
• quattro.gather()仅接受协程，不接受未来和生成器，就像TaskGroup一样。
• 当return_exceptions为假（默认值）时，子任务中的异常将导致ExceptionGroup从顶级gather()调用中冒泡出来，就像在TaskGroup中一样。
• 结果以元组的形式返回，而不是列表。

取消作用域

quattro包含了一个独立的、基于asyncio的Trio CancelScopes实现。由于asyncio和Trio之间的基本差异，实际的运行时行为并不完全相同，但很接近。

from quattro import move_on_after

async def my_handler():
    with move_on_after(1.0) as cancel_scope:
        await long_query()

    # 1 second later, the function continues running

quattro 包含以下辅助器：

• move_on_after
• move_on_at
• fail_after
• fail_at

所有辅助器都生成 quattro.CancelScope 实例，它与 Trio 变体非常相似。

CancelScopes 具有以下属性：

• cancel() - 一个可以手动取消作用域的方法。在进入作用域之前可以调用 cancel()；进入作用域时将在第一个机会取消它
• deadline - 读写，作用域的可选截止日期，在该日期作用域将被取消
• cancelled_caught - 一个只读 bool 属性，表示作用域是否通过取消完成

quattro 还支持使用 quattro.current_effective_deadline 在任务中检索当前有效截止日期。当前有效截止日期是一个浮点值，其中 float('inf') 表示没有截止日期。

Python 3.11 及更高版本包含类似的辅助器 asyncio.timeout 和 asyncio.timeout_at。quattro 的 fail_after 和 fail_at 辅助器实际上等同于 asyncio 超时，并且通过了它们的测试套件。

区别在于：

• quattro 版本是普通上下文管理器（仅使用 with），asyncio 版本是异步上下文管理器（使用 async with）。两个版本都不需要是异步的，因为没有等待；quattro 选择非异步以表示没有遇到挂起点，与 Trio 保持一致，并且更易读。
• quattro 还包含 move_on_at 和 move_on_after 辅助器。
• quattro 版本支持获取当前有效截止日期。
• quattro 版本可以使用 scope.cancel() 手动取消，并在进入之前预先取消。
• quattro 版本在所有支持的 Python 版本中均可用，而不仅仅是 3.11+。

asyncio 和 Trio 的区别：

当它们失败时，fail_after 和 fail_at 抛出 asyncio.Timeout 异常而不是 trio.Cancelled 异常。

asyncio 使用边沿触发取消语义，而 Trio 使用电平触发取消语义。以下示例在 quattro 和 Trio 中的行为将不同

with trio.move_on_after(TIMEOUT):
    conn = make_connection()
    try:
        await conn.send_hello_msg()
    finally:
        await conn.send_goodbye_msg()

在 Trio 中，如果 TIMEOUT 在等待 send_hello_msg() 时过期，则还会取消 send_goodbye_msg()。在 quattro 中，send_goodbye_msg() 仍然会运行（并可能阻塞）。这是底层框架的限制。

在 quattro 中，取消作用域无法被屏蔽。

任务组

在 Python 3.11 及更高版本上，使用标准库 TaskGroup 实现。这里的 TaskGroup 实现可以被视为旧版 Python 版本的移植。

quattro 包含一个 TaskGroup 实现。TaskGroups 受 Trio nurseries 的启发。

from quattro import TaskGroup

async def my_handler():
    # We want to spawn some tasks, and ensure they are all handled before we return.
    async def task_1():
        ...

    async def task_2():
        ...

    async with TaskGroup() as tg:
        t1 = tg.create_task(task_1)
        t2 = tg.create_task(task_2)

    # The end of the `async with` block awaits the tasks, ensuring they are handled.

TaskGroups 是实现 结构化并发 的基本构建块。简单来说，结构化并发意味着你的代码不会泄漏任务 - 当协程完成时，该协程生成的所有任务及其所有子任务也将完成。（用时髦的话说，执行流程成为一个有向无环图。）

可以使用 TaskGroups 代替 asyncio.create_task 来启动后台任务来实现结构化并发。TaskGroups 实际上做了两件事

• 当退出 TaskGroup 的 async with 块时，TaskGroup 等待所有子任务，确保它们在退出时已完成
• 当 TaskGroup 子任务引发异常时，所有其他子任务和上下文管理器内的任务将被取消

实现已从EdgeDB项目借用。

变更日志

24.1.0 (2024-05-01)

• 添加Trove分类器。
• 将name关键字参数添加到TaskGroup.create_task。(#8)

23.1.0 (2023-11-29)

• 引入quattro.gather。(#5)
• 添加对Python 3.12的支持。
• 切换到PDM。

22.2.0 (2022-12-27)

• 3.11上的嵌套取消操作更健壮。
• 对fail_after和fail_at提供更好的类型支持。
• 改进已取消作用域的截止日期处理。
• TaskGroups现在在创建任务时支持自定义ContextVar上下文，就像标准库实现一样。

22.1.0 (2022-12-19)

• 恢复TaskGroup版权声明。
• TaskGroups现在在子错误上抛出ExceptionGroups（在必要时使用PyPI回滚版）。
• 添加对Python 3.11的支持，弃用3.8。
• TaskGroups不再有name，且repr略有不同，以与Python 3.11标准库实现保持一致。
• TaskGroups在终止时不再吞并子异常，以与Python 3.11标准库实现保持一致。
• 切换到CalVer。

0.3.0 (2022-01-08)

• 添加py.typed以启用类型信息。
• 完善TaskGroups的类型注释。

0.2.0 (2021-12-27)

• 添加quattro.current_effective_deadline。

0.1.0 (2021-12-08)

• 首次发布，包含任务组和取消作用域。

致谢

初始TaskGroup实现来自EdgeDB项目。CancelScope实现受到Trio的极大影响，并受到async_timeout包的启发。