时间戳

首页 > 解决方案 > 如何将异步迭代器视为同步迭代器？

python - 如何将异步迭代器视为同步迭代器？

问题描述

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

当然，你可以使用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]

标签： pythonpython-asyncio

解决方案

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())

但是后面的代码仅作为学习练习，不要尝试在生产中使用它。

推荐阅读