在异步生成器上使用 next()

Question

next()可以使用内置函数逐步迭代生成器。例如：

def sync_gen(n):
    """Simple generator"""
    for i in range(n):
        yield i**2

sg = sync_gen(4)
print(next(sg)) # -> 0
print(next(sg)) # -> 1
print(next(sg)) # -> 4

在异步生成器上使用next()不起作用：

import asyncio

async def async_gen(n):
    for i in range(n):
        yield i**2

async def main():
    print("Async for:")
    async for v in async_gen(4):  # works as expected
        print(v)

    print("Async next:")
    ag = async_gen(4)
    v = await next(ag) # raises: TypeError: 'async_generator' object is not an iterator
    print(v)

asyncio.run(main())

是否存在类似的东西v = await async_next(ag)来获得与普通生成器相同的行为？

Answer 1

从Python 3.10开始，有了aiter(async_iterable)和awaitable anext(async_iterator)内置函数，类似于iter和next，所以你不必再依赖async_iterator.__anext__()魔术方法了。这段代码适用于 python 3.10：

import asyncio


async def async_gen(n):
    for i in range(n):
        yield i**2


async def main():
    print("Async next:")
    ag = async_gen(4)
    print(await anext(ag))


asyncio.run(main())

Answer 2

如前所述，next它不适用于异步生成器，因为它不是迭代器。

如果必须使用函数，您可以创建一个辅助函数，而不是使用__anext__函数 outwrite：

async def anext(ait):
    return await ait.__anext__()

ag = async_gen(4)
v = await anext(ag)
print(v)

编辑：

从 python 3.10 开始，他们将其作为内置函数引入：anext