Python anext() 函数

Python anext() 语法

anext() 是 Python 3.10 引入的一个内置函数，它用于获取异步迭代器的下一个元素。

语法：

await anext(async_iterator, default)

说明：

anext() 函数接收以下 2 个参数。

async_iterator（必选）：是一个异步迭代器对象（需通过 aiter() 获取）。
default（可选）：迭代耗尽时返回的默认值。如果 default 省略，则耗尽时会引发 StopAsyncIteration 异常。

anext() 函数会返回异步迭代器的下一个元素（需要通过 await 获取实际值）。若迭代器耗尽且提供 default 参数，则返回 default 值。

Python anext() 摘要

使用频率	中
最低版本	Python 3.10
官方文档	查看
相关函数	aiter()、iter()、next()

Python anext() 示例

接下来，我们通过几个简单的例子来讲解一下 Python anext() 函数是如何使用的。

注意: 绿叶网的 Python 在线编译器只会一次性返回所有结果。对于本节所有示例，必须在本地运行才会有异步效果。

示例 1：anext() 基本用法

import asyncio

async def async_gen():
    for char in 'ABC':
        await asyncio.sleep(1)    # 模拟异步操作
        yield char

async def main():
    ait = aiter(async_gen())    # 获取异步迭代器
    print(await anext(ait))     # A
    print(await anext(ait))     # B
    print(await anext(ait))     # C
    print(await anext(ait))     # 触发 StopAsyncIteration

asyncio.run(main())      # 输出 A → B → C → 异常

运行结果如下。

A
B
C
（异常）

分析：

上面定义了一个异步生成器 async_gen()，它每隔 1 秒依次生成字符 'A'、'B' 和 'C'。这里每次生成之间会延迟 1 秒，主要是为了模拟异步操作。

在 main() 函数中，使用 aiter() 获取异步迭代器，并通过 anext() 逐个获取数据；当数据流耗尽时，anext() 会触发 StopAsyncIteration 异常。最终，asyncio.run(main()) 运行异步程序，依次输出 A、B、C，并在尝试获取下一个值时抛出异常。

示例 2：anext() 带默认值

import asyncio

async def paginated_fetch():
    pages = ['Page1', 'Page2']
    for page in pages:
        await asyncio.sleep(2)
        yield page

async def main():
    ait = aiter(paginated_fetch())
    print(await anext(ait, 'No Data'))      # Page1
    print(await anext(ait, 'No Data'))      # Page2
    print(await anext(ait, 'No Data'))      # No Data（返回默认值）
    print(await anext(ait, 'No Data'))      # 仍返回默认值

asyncio.run(main())

运行结果如下。

Page1
Page2
No Data
No Data

分析：

在上面代码中，首先定义了一个异步生成器 paginated_fetch()，它会每隔 2 秒依次生成 ['Page1', 'Page2']。这里每次生成之间延迟 2 秒，主要是为了模拟异步操作。

在 main() 函数中，使用 aiter() 获取异步迭代器，并通过 anext() 逐个获取数据；当数据流耗尽时，anext() 会返回默认值 'No Data'。最终，asyncio.run(main()) 运行异步程序，依次输出 Page1、Page2，并在尝试获取更多值时返回 'No Data'。

Python anext() 注意事项

在使用 anext() 函数时，小伙伴们需要特别注意以下几点。

1. 必须搭配 await 关键字使用

如果忘记使用 await 关键字，则会导致返回协程对象而非实际值，比如：

# 错误用法
print(anext(ait))      # 输出协程对象 <coroutine object...>

2. 迭代器验证

如果对非异步迭代器使用 anext() 函数，则会触发 TypeError 异常，比如：

sync_iter = iter([1,2,3])
await anext(sync_iter)      # TypeError: 'list_iterator' is not async iterable