come posso mappare/filtrare asincronicamente un iterabile asincrono?

Question

Diciamo che ho un iterabile asincrono che posso passare usando async for, come posso quindi mapparlo e filtrarlo in un nuovo iteratore asincrono? Il seguente codice, che è un adattamento di come farei la stessa cosa con un iterabile sincrono, non funziona, dal momento che yield non è consentito all'interno di async def s.

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) 

Answer 1

A recently published PEP draft (PEP 525), il cui sostegno è scheduled for Python 3.6, si propone di consentire generatori asincroni con la stessa sintassi si è venuto con.

Nel frattempo, è possibile utilizzare la libreria asyncio_extras menzionata da CryingCyclops nel relativo commento se non si desidera gestire il boiler iteratore asincrono.

Da the docs:

@async_generator 
async def mygenerator(websites): 
    for website in websites: 
     page = await http_fetch(website) 
     await yield_async(page) 

async def fetch_pages(): 
    websites = ('http://foo.bar', 'http://example.org') 
    async for sanitized_page in mygenerator(websites): 
     print(sanitized_page) 

V'è anche il async_generator library che supporta yield from costrutti.

Answer 2

can't consumo all'interno di coroutine. Per implementare la tua idea, l'unico modo che vedo è implementare Asynchronous Iterator. Se ho ragione, qualcosa del genere:

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) 

Proviamoci. Ecco esempio completo di aiuto arange classe (l'ho preso da 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()) 

uscita:

6 
8