Add asyncio.Queue __aiter__, __anext__ methods

[RekGRpth]

BPO	28777
Nosy	@gvanrossum, @cjerdonek, @1st1, @RekGRpth

^{Note: these values reflect the state of the issue at the time it was migrated and might not reflect the current state.}

Show more details

GitHub fields:

assignee = None
closed_at = <Date 2017-08-18.20:27:30.503>
created_at = <Date 2016-11-23.04:58:52.383>
labels = ['type-feature', '3.7', 'expert-asyncio']
title = 'Add asyncio.Queue __aiter__, __anext__ methods'
updated_at = <Date 2017-08-18.20:27:30.502>
user = 'https://github.com/RekGRpth'

bugs.python.org fields:

activity = <Date 2017-08-18.20:27:30.502>
actor = 'gvanrossum'
assignee = 'none'
closed = True
closed_date = <Date 2017-08-18.20:27:30.503>
closer = 'gvanrossum'
components = ['asyncio']
creation = <Date 2016-11-23.04:58:52.383>
creator = 'RekGRpth'
dependencies = []
files = []
hgrepos = []
issue_num = 28777
keywords = []
message_count = 9.0
messages = ['281536', '281749', '281750', '299131', '299200', '299688', '299714', '300496', '300530']
nosy_count = 5.0
nosy_names = ['gvanrossum', 'chris.jerdonek', 'yselivanov', 'RekGRpth', 'Matt Rasband']
pr_nums = []
priority = 'normal'
resolution = 'rejected'
stage = 'resolved'
status = 'closed'
superseder = None
type = 'enhancement'
url = 'https://bugs.python.org/issue28777'
versions = ['Python 3.7']

[RekGRpth]

adding to asyncio.Queue class following methods:
def __aiter__(self): return self
async def __anext__(self): return await self.get()
let use asyncio.Queue follow:
queue = asyncio.Queue()
...
async for item in queue: do_something_with(item)

[gvanrossum]

This should be an upstream PR first (GitHub.com/python/asyncio). Also, too late for 3.6.

[gvanrossum]

Also it's not clear that it's a good idea without more thought -- there's no way to end the loop on the producing side.

[cjerdonek]

there's no way to end the loop on the producing side.

I might be missing something, but can't something similar be said of queue.get()?

[gvanrossum]

> there's no way to end the loop on the producing side.

I might be missing something, but can't something similar be said of queue.get()?

That's my point, actually. If you are wrapping the Queue protocol with __aiter__/anext the caller would expect there's a way to signal to the latter that the loop is over (so it can raise StopAsyncIteration). But since .get() doesn't have a way to signal this, an async for-loop would not be able to terminate (other than through break).

[RekGRpth]

I successfully use my code:

import asyncio, sanic

class MyQueue(asyncio.Queue):
    def __aiter__(self): return self
    async def __anext__(self): return await self.get()

app = sanic.Sanic()
ws_set = set()
app.static('/', 'async.html')

@app.websocket('/ws')
async def root_ws(request, ws):
ws_set.add(ws)
try:
while True: await ws.recv()
finally: ws_set.remove(ws)

async def postgres():
    import aiopg
    async with aiopg.create_pool('') as pool:
        async with pool.acquire() as connection:
            connection._notifies = MyQueue()
            async with connection.cursor() as cursor:
                await cursor.execute('LISTEN message')
                async for message in connection.notifies:
                    for ws in ws_set: await ws.send(message.payload)

try: asyncio.get_event_loop().run_until_complete(asyncio.gather(app.create_server(), postgres()))
except KeyboardInterrupt: asyncio.get_event_loop().stop()

[gvanrossum]

So that's an infinite loop right?

[MattRasband]

I attempted this myself, it seemed to have too many costs associated for the stdlib and is something easy enough to wrap myself when I need this functionality with explicit semantics on how to "stop" the queue (using an object() sentinel). My implementation is flawed as the sentinel is global, instead of per queue instance, if you decide to use this at all.

See python/asyncio#445

[gvanrossum]

Let's not do this.