寻找PyTest中适配异步代码的Wait-Until/轮询工具库

嗨，我特别理解你这种心情——自己写个轮询函数虽然不难，但现成的库不仅省心，还能帮你处理好多你可能没考虑到的边界情况，比如异常捕获、超时处理什么的。给你推荐几个适合异步PyTest场景的工具库，完全能实现你想要的await_until类似功能：

• tenacity：这个库主打重试逻辑，但灵活性超强，稍微配置下就能完美当轮询等待工具用，而且原生支持异步。你可以直接封装成你想要的await_until风格，用法和你预想的几乎一模一样：

from tenacity import retry, stop_after_delay, wait_fixed, retry_if_result
import asyncio

async def await_until(condition_func, interval=1, timeout=10):
    @retry(
        stop=stop_after_delay(timeout),  # 超时10秒就停止
        wait=wait_fixed(interval),       # 每次检查间隔1秒
        retry=retry_if_result(lambda res: not res)  # 条件不满足就重试
    )
    async def check_condition():
        # 兼容同步和异步的条件函数
        result = condition_func()
        if asyncio.iscoroutine(result):
            return await result
        return result
    await check_condition()

# 用的时候就和你想要的一样
operation.run_in_background()
await await_until(lambda: operation.is_ready(), interval=1, timeout=10)

• aioretry：这是专门为异步场景设计的重试/轮询库，语法更简洁，不需要太多配置就能实现需求：

from aioretry import retry

async def await_until(condition_func, interval=1, timeout=10):
    max_tries = int(timeout / interval) + 1  # 计算最大尝试次数
    @retry(tries=max_tries, delay=interval)
    async def check_condition():
        # 如果你的is_ready是同步方法，直接调用就行，不用await
        if not await condition_func():
            raise AssertionError("还没准备好，继续轮询")
    await check_condition()

# 使用示例
operation.run_in_background()
await await_until(lambda: operation.is_ready(), interval=1, timeout=10)

要是你暂时不想加新的依赖，其实用Python标准库的asyncio也能快速写个极简版，不过既然你要现成的，上面这两个库都是社区里常用的，稳定性和维护性都有保障。

备注：内容来源于stack exchange，提问作者Aleks Ya