如何像使用同步迭代器一样使用异步迭代器并适配自定义列表类

开发者阿泽

2026-10-3

问题原因

Python 内置list的构造函数仅支持接收同步可迭代对象，你传入的异步生成器属于异步可迭代对象，同步逻辑无法直接消费，因此直接调用MyList(x async for x in iterate())会报错。

最优实现方案

给自定义MyList添加异步构造类方法，直接迭代异步对象填充数据，全程没有中间列表生成，无额外内存开销：

import asyncio
from typing import AsyncIterable, Any


class MyList(list):
    @classmethod
    async def afrom_async_iter(cls, async_iter: AsyncIterable[Any]) -> "MyList":
        instance = cls()
        async for item in async_iter:
            instance.append(item)
        return instance


async def iterate():
    yield 1
    yield 2


async def main():
    # 仅需await调用异步构造方法即可，调用逻辑简洁
    return await MyList.afrom_async_iter(iterate())


result = asyncio.run(main())
print(result) # 输出 [1, 2]
print(type(result)) # 输出 <class '__main__.MyList'>

如果你不想修改MyList类的实现，也可以单独封装异步工具函数实现相同效果：

async def build_mylist(async_iter: AsyncIterable[Any]) -> MyList:
    res = MyList()
    async for item in async_iter:
        res.append(item)
    return res

# 调用方式
async def main():
    return await build_mylist(iterate())