如何像对待同步迭代器一样对待异步迭代器？

Question

理解符号支持从异步迭代器生成列表。我如何在自定义列表中执行此操作 class?

当然可以使用list.append生成列表，但我想知道聪明的方法。

import asyncio


class MyList(list):
    pass


async def iterate():
    yield 1
    yield 2


async def main():
    return [x async for x in iterate()]
    # return MyList(x async for x in iterate()) # NG
    # return MyList([x async for x in iterate()])  # OK However, there is a lot of waste because the list is generated twice

result = asyncio.run(main())
print(result) # [1, 2]

Answer 1

Python 魔术方法不是异步的 - 例如，在您的 main() 中，您永远不会 await MyList 构造函数，因此它也不能是异步的.通常的方法是提供一个方法，例如 extend_async 并使用它：

class MyList(list):
    async def extend_async(self, aiter):
        async for obj in aiter:
            self.append(obj)

async def main():
     lst = MyList()
     await lst.extend_async(x async for x in iterate())
     return lst

或者，如果你想让 main 更短，你可以创建一个助手：

class MyList(list):
    @staticmethod
    async def from_aiter(aiter):
        self = MyList()
        async for obj in aiter:
            self.append(obj)
        return self

async def main():
     return await MyList.from_aiter(x async for x in iterate())

最后，说了那么多......如果你真的坚持异步构造，你可以使用元类破解它:

class Hack(type):
    async def __call__(cls, aiter):
        self = type.__call__(cls)
        await self.extend_async(aiter)
        return self

class MyList(list, metaclass=Hack):
    async def extend_async(self, aiter):
        async for obj in aiter:
            self.append(obj)

async def main():
     return await MyList(x async for x in iterate())

但后一个代码仅作为学习练习，请勿尝试在生产中使用它。

如何像对待同步迭代器一样对待异步迭代器？

How can I treat an asynchronous iterator like a sync iterator?

python

python-asyncio