Python内置函数aiter()的作用及具体使用方法是什么？

基本定义 #

aiter(async_iterable)是Python3.10新增的内置函数，作用是为异步可迭代对象返回对应的异步迭代器，和同步场景下的iter()函数功能对应。
其底层逻辑是调用传入对象的__aiter__()魔术方法，返回的异步迭代器可以配合anext()函数或者async for语法完成遍历。

常见适用场景 #

• 手动控制异步迭代流程：不需要全量遍历所有元素、或者需要在迭代中途插入自定义逻辑的场景
• 异步迭代器二次封装：给原有异步可迭代对象添加限流、重试、日志埋点、过滤等装饰逻辑
• 统一异步迭代调用规范：避免直接调用__aiter__()魔术方法的不规范写法，兼容不同实现的异步可迭代对象

可运行代码示例 #

示例1：基础用法对比 #

以下代码实现了一个异步计数器，分别演示async for隐式调用aiter和手动调用aiter控制迭代的效果：

import asyncio
from typing import AsyncIterable, AsyncIterator

# 自定义异步可迭代对象：异步计数器
class AsyncCounter:
    def __init__(self, start: int, end: int):
        self.current = start
        self.end = end
    
    def __aiter__(self) -> AsyncIterator[int]:
        return self
    
    async def __anext__(self) -> int:
        if self.current >= self.end:
            raise StopAsyncIteration
        await asyncio.sleep(0.1) # 模拟异步IO延迟
        self.current += 1
        return self.current - 1

# 方式1：async for 隐式调用aiter完成遍历
async def demo_auto_iter():
    print("=== 自动迭代（async for） ===")
    async for num in AsyncCounter(0, 3):
        print(num)

# 方式2：手动调用aiter控制迭代流程
async def demo_manual_iter():
    print("\n=== 手动控制迭代 ===")
    counter = AsyncCounter(0, 3)
    # 显式调用aiter获取异步迭代器
    iter_obj = aiter(counter)
    # 手动获取第一个元素
    first_item = await anext(iter_obj)
    print(f"手动取出第一个元素：{first_item}")
    # 迭代中途插入自定义业务逻辑
    print("执行其他异步操作...")
    await asyncio.sleep(0.2)
    # 继续遍历剩余元素
    async for num in iter_obj:
        print(f"剩余元素：{num}")

if __name__ == "__main__":
    asyncio.run(demo_auto_iter())
    asyncio.run(demo_manual_iter())

运行输出：

=== 自动迭代（async for） ===
0
1
2

=== 手动控制迭代 ===
手动取出第一个元素：0
执行其他异步操作...
剩余元素：1
剩余元素：2

示例2：异步迭代封装 #

以下代码基于aiter实现异步迭代的限流装饰器，控制每秒最多输出2个元素：

def limit_async_iter(aiterable: AsyncIterable, max_per_second: int):
    # 用aiter统一把输入转成异步迭代器
    base_iter = aiter(aiterable)
    interval = 1 / max_per_second
    async def wrapper():
        while True:
            try:
                start_time = asyncio.get_event_loop().time()
                item = await anext(base_iter)
                yield item
                # 补全等待时间实现限流
                wait_time = interval - (asyncio.get_event_loop().time() - start_time)
                if wait_time > 0:
                    await asyncio.sleep(wait_time)
            except StopAsyncIteration:
                break
    return wrapper()

async def demo_limit_iter():
    print("\n=== 限流异步迭代 ===")
    counter = AsyncCounter(0, 5)
    # 封装成每秒最多输出2个的异步迭代器
    limited_iter = limit_async_iter(counter, max_per_second=2)
    async for num in limited_iter:
        print(f"输出元素：{num}，时间戳：{asyncio.get_event_loop().time():.2f}")

if __name__ == "__main__":
    asyncio.run(demo_limit_iter())

注意事项 #

• aiter只能传入实现了__aiter__方法的异步可迭代对象，传入列表、生成器等同步可迭代对象会抛出TypeError
• 普通全量遍历场景直接使用async for即可，不需要显式调用aiter
• aiter通常和anext配对使用，对应同步场景的iter和next函数

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