Python中如何在生成器表达式中搭配any使用async_generator

Python内置的any()、all()都是同步函数，仅支持遍历同步可迭代对象。你写的(x async for x in foo.iter())是异步生成器表达式，属于异步可迭代范畴，同步any()没有能力驱动它的异步迭代逻辑，因此直接抛出TypeError: 'async_generator' object is not iterable异常。

你之前跑通的any([x async for x in async_generator])写法，本质是先通过异步列表推导把异步生成器的所有元素一次性全量加载到内存，拼成同步列表后再传给同步any()。这种写法存在明显缺陷：

• 完全丢失生成器惰性求值的优势，如果异步生成器数据量极大、或是无限流，会直接占满内存
• 必须迭代完所有元素才能得到结果，没法像原生生成器那样遇到第一个符合条件的元素就提前终止迭代
• 该写法本身也只能在async def定义的协程函数内运行，无法在同步上下文执行

自己实现一个异步版本的any即可，行为和内置同步any()完全一致，保留惰性求值特性，遍历到第一个真值元素就立刻返回，不需要走完整个异步迭代流程：

async def async_any(async_iter):
    async for item in async_iter:
        if item:
            return True
    return False

使用方式和同步版几乎一致，调用时加await即可，同样支持带条件判断的异步生成器表达式：

async def check_dir_status():
    # 直接判断异步迭代器是否存在真值元素
    not_empty = await async_any(foo.iter())
    # 支持自定义判断逻辑，和原生any的生成器写法对齐
    has_py_file = await async_any(item.suffix == '.py' async for item in foo.iter())

如果不想自行维护工具函数，也可以直接使用成熟的异步迭代工具库提供的对应实现，行为和上述自写函数完全一致。

补充说明：你给出的第一个同步示例存在笔误，Pathlib应改为Path，正确写法为path = Path("/foo/bar/baz")。

内容的提问来源于stack exchange，提问作者Batman