ZetCode

Python aiter 函数

上次修改时间：2025 年 4 月 11 日

这篇综合指南探讨了 Python 的 aiter 函数，该函数返回一个用于异步迭代的异步迭代器。我们将涵盖基本用法、自定义异步迭代器和实际示例。

基本定义

aiter 函数返回一个异步迭代器对象。 它是内置 iter 函数的异步等效项。 它在 Python 3.10 中引入，适用于实现 __aiter__ 的对象。

主要特征：在 async for 循环中使用，需要一个异步可迭代对象，返回一个异步迭代器。 它通常与 anext 一起用于异步迭代。

基本异步迭代

这是一个使用异步生成器的简单示例，展示了 aiter 如何与基本异步迭代一起工作。

basic_aiter.py 
async def async_gen():
    for i in range(3):
        yield i
        await asyncio.sleep(0.1)

async def main():
    ag = async_gen()
    ait = aiter(ag)
    
    while True:
        try:
            val = await anext(ait)
            print(val)
        except StopAsyncIteration:
            break

asyncio.run(main())

此示例显示了基本异步迭代。 async_gen 异步生成值。 aiter 获取异步迭代器，anext 检索值。

循环持续进行，直到引发 StopAsyncIteration，类似于常规迭代，但使用 async/await 语法。

自定义异步迭代器

您可以通过实现 __aiter__ 来创建自定义异步迭代器。 此示例显示了一个简单的异步计数器。

custom_aiter.py 
class AsyncCounter:
    def __init__(self, stop):
        self.stop = stop
        self.current = 0
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.current >= self.stop:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)
        self.current += 1
        return self.current - 1

async def main():
    async for i in AsyncCounter(3):
        print(i)

asyncio.run(main())

AsyncCounter 类实现了 __aiter__ 和 __anext__ 以支持异步迭代。 调用 aiter 时，它将返回此对象。

当您需要比异步生成器提供更多地控制异步迭代时，此模式非常有用。

异步可迭代协议

此示例演示了具有单独迭代器和可迭代类的完整异步可迭代协议。

protocol.py 
class AsyncIterable:
    def __init__(self, data):
        self.data = data
    
    def __aiter__(self):
        return AsyncIterator(self.data)

class AsyncIterator:
    def __init__(self, data):
        self.data = data
        self.index = 0
    
    async def __anext__(self):
        if self.index >= len(self.data):
            raise StopAsyncIteration
        await asyncio.sleep(0.1)
        item = self.data[self.index]
        self.index += 1
        return item

async def main():
    ait = aiter(AsyncIterable([1, 2, 3]))
    print(await anext(ait))  # 1
    print(await anext(ait))  # 2
    print(await anext(ait))  # 3

asyncio.run(main())

这显示了具有单独可迭代和迭代器类的完整协议。 AsyncIterable 返回一个新的 AsyncIterator。

这种分离允许对相同数据进行多次独立迭代，类似于常规可迭代对象的工作方式。

错误处理

当与非异步可迭代对象一起使用时，aiter 函数会引发 TypeError。 此示例显示了正确的错误处理。

errors.py 
async def main():
    try:
        ait = aiter([1, 2, 3])  # Regular list is not async iterable
    except TypeError as e:
        print(f"Error: {e}")  # 'list' object is not async iterable

    class NoAiter:
        pass
    
    try:
        ait = aiter(NoAiter())
    except TypeError as e:
        print(f"Error: {e}")  # 'NoAiter' object is not async iterable

asyncio.run(main())

这些示例演示了 aiter 在无效类型上的行为。 常规可迭代对象和没有 __aiter__ 的对象会引发错误。

要使一个类与 aiter 一起使用，请如前面的示例所示实现 __aiter__。

实际示例：异步数据获取

此示例显示了 aiter 的实际用例，用于从多个源进行异步数据获取。

data_fetch.py 
async def fetch_data(url):
    # Simulate network request
    await asyncio.sleep(0.2)
    return f"Data from {url}"

class AsyncDataFetcher:
    def __init__(self, urls):
        self.urls = urls
    
    def __aiter__(self):
        self.index = 0
        return self
    
    async def __anext__(self):
        if self.index >= len(self.urls):
            raise StopAsyncIteration
        url = self.urls[self.index]
        self.index += 1
        return await fetch_data(url)

async def main():
    fetcher = AsyncDataFetcher([
        "api.example.com/1",
        "api.example.com/2",
        "api.example.com/3"
    ])
    
    async for data in fetcher:
        print(data)

asyncio.run(main())

这演示了一个真实世界的用例，其中 aiter 启用了异步数据处理。 获取器逐个从 URL 检索数据。

异步迭代模式允许有效的 I/O 绑定操作，而不会阻塞事件循环。

最佳实践

资料来源

作者

我叫 Jan Bodnar，是一位充满激情的程序员，拥有丰富的编程经验。 自 2007 年以来，我一直在撰写编程文章。 迄今为止，我撰写了 1,400 多篇文章和 8 本电子书。 我拥有超过十年的编程教学经验。

列出所有 Python 教程。