如何在Python中实现自定义的异步迭代器接口？-Python教程-PHP中文网

异步迭代器的__aiter__必须是同步方法且返回自身（或具备__anext__的对象），因为async for会严格检查其返回值是否含__anext__；__anext__必须是async def，返回awaitable。

如何在python中实现自定义的异步迭代器接口？

Python 的异步迭代器必须同时实现 __aiter__ 和 __anext__，缺一不可；只加 async def __anext__ 而不返回异步迭代器对象，会直接报 TypeError: 'X' object is not an async iterator。

为什么 `aiter` 必须返回自身（且是异步迭代器）？

Python 运行时在遇到 async for 时，会先调用 __aiter__，并**严格检查其返回值是否具备 __anext__ 方法**。它不会自动包装、也不接受协程对象。

常见错误写法：async def __aiter__(self): return self —— 这是语法错误，__aiter__ 不能是 async 的；正确做法是让它同步返回一个实现了 __anext__ 的对象（通常是 self）。

__aiter__ 必须是普通方法（不能加 async），返回值必须有 __anext__ 方法
__anext__ 必须是 async def，返回一个 awaitable（通常用 return 或 raise StopAsyncIteration）
若把耗时操作（如网络请求、文件读取）放在 __anext__ 中，每次 async for 迭代都会真正 await 它

一个带状态的异步迭代器示例：从 API 分页拉取数据

典型场景：你封装一个异步迭代器，每次 __anext__ 触发一次 HTTP 请求，自动处理分页逻辑。

import asyncio
import aiohttp
<p>class AsyncPageIterator:
def <strong>init</strong>(self, base_url: str, page_size: int = 10):
self.base_url = base_url
self.page_size = page_size
self._page = 1
self._session = None  # 注意：session 不应在 <strong>init</strong> 中创建</p><pre class="brush:php;toolbar:false;">def __aiter__(self):
    return self

async def __anext__(self):
    if self._session is None:
        # 延迟到首次 __anext__ 再初始化 session，避免在 __aiter__ 中做 async 操作
        self._session = aiohttp.ClientSession()

    url = f"{self.base_url}?page={self._page}&size={self.page_size}"
    try:
        async with self._session.get(url) as resp:
            data = await resp.json()
            if not data:
                raise StopAsyncIteration
            self._page += 1
            return data
    except Exception:
        await self._session.close()
        raise StopAsyncIteration

macOS UI 自动化技能包 —— 用 Python 控制你的 Mac

让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限

下载

注意点：

不要在 __aiter__ 中创建 aiohttp.ClientSession，它不是线程安全，也不能跨事件循环复用
异常后记得 await self._session.close()，否则连接泄漏
真实项目中建议用 contextlib.AsyncExitStack 管理资源，但基础迭代器里手动处理更直观

容易被忽略的兼容性陷阱

异步迭代器和普通迭代器行为不一致，尤其在边界和调试时表现不同：

async for x in it: 无法像普通 for 那样用 next(it) 手动触发；必须用 await it.__anext__()
没有内置的 aiter() / anext() 函数（Python 3.10+ 才有 anext()，但无 aiter()）
不能混用：给 async for 传一个只实现 __iter__/__next__ 的对象，会报 TypeError: 'X' object is not an async iterator
若想支持同步和异步两种遍历，需分别实现两套接口（即同时有 __iter__+__next__ 和 __aiter__+__anext__）