我怎么能异步映射/过滤异步迭代？ [英] how can I asynchronously map/filter an asynchronous iterable?

问题描述

让我们说我有一个异步迭代，我可以越过使用异步为，怎能我然后映射，并筛选出一个新的异步迭代器？下面code这是自收益不允许在<$我怎么会做同样的事情，与同步迭代不工作，适应C $ C>异步DEF 秒。

 异步高清mapfilter（aiterable，P，FUNC）：
    异步在aiterable有效载荷：
        如果p（负载）：            ＃这部分是不允许的，但希望它应该清楚
            ＃我想要完成的任务。
            产量FUNC（负载）
 

解决方案

您的无法的协同程序内使用产量。为了实现你的想法，只是我看不到出路是实施的异步迭代器的。如果我是正确的，这样的事情：

 类MapFilter：
    高清__init __（自我，aiterable，P，FUNC）：
        self.aiterable = aiterable
        self.p = P
        self.func = FUNC    异步高清__aiter __（个体经营）：
        自回归    异步高清__anext __（个体经营）：
        而真正的：
            有效载荷=的await self.aiterable .__ ANEXT __（）＃StopAsyncIteration将在这里养上没有新的价值观
            如果self.p（负载）：
                返回self.func（负载）
 

让我们来测试它。下面是与助手完整的例子人气指数类（我把它从这里 ）：

 进口ASYNCIO
类人气指数：
    高清__init __（自我，N）：
        self.n = N
        self.i = 0    异步高清__aiter __（个体经营）：
        自回归    异步高清__anext __（个体经营）：
        I = self.i
        self.i + = 1
        如果self.i＆LT; = self.n：
            等待asyncio.sleep（0）＃插入屈服点
            我返回
        其他：
            提高StopAsyncIteration
类MapFilter：
    高清__init __（自我，aiterable，P，FUNC）：
        self.aiterable = aiterable
        self.p = P
        self.func = FUNC    异步高清__aiter __（个体经营）：
        自回归    异步高清__anext __（个体经营）：
        而真正的：
            有效载荷=的await self.aiterable .__ ANEXT __（）
            如果self.p（负载）：
                返回self.func（负载）
异步高清的main（）：
    aiterable =人气指数（5）
    p值=拉姆达X：布尔（X→2）
    FUNC =拉姆达X：X * 2    在MapFilter（aiterable，P，FUNC）为异步I：
        打印（一）如果__name__ ==__main__：
    循环= asyncio.get_event_loop（）
    loop.run_until_complete（主（））
 

输出：

  6
8
 

Let's say I have an asynchronous iterable that I can pass over using async for, how then can I then map and filter it to a new asynchronous iterator? The following code which is an adaptation of how I'd do the same thing with a synchronous iterable doesn't work, since yield isn't allowed inside async defs.

async def mapfilter(aiterable, p, func):
    async for payload in aiterable:
        if p(payload):

            # This part isn't allowed, but hopefully it should be clear
            # what I'm trying to accomplish.
            yield func(payload)

解决方案

You can't use yield inside coroutines. To implement your idea, only way I see is to implement Asynchronous Iterator. If I'm right, something like that:

class MapFilter:
    def __init__(self, aiterable, p, func):
        self.aiterable = aiterable
        self.p = p
        self.func = func

    async def __aiter__(self):
        return self

    async def __anext__(self):
        while True:
            payload = await self.aiterable.__anext__()  # StopAsyncIteration would be raise here on no new values
            if self.p(payload):
                return self.func(payload)

Let's test it. Here's complete example with helper arange class (I took it from here):

import asyncio


class arange:
    def __init__(self, n):
        self.n = n
        self.i = 0

    async def __aiter__(self):
        return self

    async def __anext__(self):
        i = self.i
        self.i += 1
        if self.i <= self.n:
            await asyncio.sleep(0)  # insert yield point
            return i
        else:
            raise StopAsyncIteration


class MapFilter:
    def __init__(self, aiterable, p, func):
        self.aiterable = aiterable
        self.p = p
        self.func = func

    async def __aiter__(self):
        return self

    async def __anext__(self):
        while True:
            payload = await self.aiterable.__anext__()
            if self.p(payload):
                return self.func(payload)


async def main():
    aiterable = arange(5)
    p = lambda x: bool(x>2)
    func = lambda x: x*2

    async for i in MapFilter(aiterable, p, func):
        print(i)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Output:

6
8

这篇关于我怎么能异步映射/过滤异步迭代？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！