Get all items in an asyncio.Queue and return them

Question

How do you gather all items from an asyncio.Queue instance and return them as a result?

Caveats:

The total number of items put into the queue cannot be known ahead of time (but is finite, and will fit into memory).
Multiple producers can add items to the queue; the producers themselves don't necessarily know when all items have been added to the queue.

The examples I have seen for consumers of an asyncio.Queue have not had to gather and return the results; they use the item from the queue but have no return value. They rely on side-effects to do their work, and don't care about returning a result.

To be concrete, below is a simplistic example. The only way I figured out how to make this work is to provide an output parameter/output argument called items to the coroutine gathering the results from the queue:

import asyncio
import random


async def add_queue_item(item, queue):
    # simulate some work
    sleep_interval = random.randint(0, 3)
    await asyncio.sleep(sleep_interval)
    output_item = item + 1
    await queue.put(output_item)


async def get_all_queue_items(queue, items):
    while True:
        items.append(await queue.get())
        queue.task_done()


async def main():
    queue = asyncio.Queue()
    items = []
    producer_tasks = [asyncio.create_task(add_queue_item(item, queue)) for item in range(5)]
    collect_queue_items_task = asyncio.create_task(get_all_queue_items(queue, items))
    await queue.join()
    await asyncio.gather(*producer_tasks)
    collect_queue_items_task.cancel()
    print(items)
    assert sorted(items) == [1, 2, 3, 4, 5]


asyncio.run(main())

Is there a way to implement get_all_queue_items above such that we can await to get all the items — making it clear what's intended? i.e.,

    …
    await queue.join()
    await asyncio.gather(*producer_tasks)
    items = await 
    print(items)
    assert sorted(items) == [1, 2, 3, 4, 5]

gotgenes_38140 · Answer

I was able to get an implementation using a sentinel value to alert the consumer get_all_queue_items there are no more values to expect from the queue, breaking it out of its loop. The task that schedules get_all_queue_items can be awaited and will have the collected items in it.

import asyncio
import random


SENTINEL = object()


async def add_queue_item(item, queue):
    # simulate some work
    sleep_interval = random.randint(1, 3)
    await asyncio.sleep(sleep_interval)
    output_item = item + 1
    await queue.put(output_item)


async def get_all_queue_items(queue):
    items = []
    item = await queue.get()
    while item is not SENTINEL:
        items.append(item)
        queue.task_done()
        item = await queue.get()
    queue.task_done()
    return items


async def main():
    queue = asyncio.Queue()
    producer_tasks = [asyncio.create_task(add_queue_item(item, queue)) for item in range(5)]
    collect_queue_items_task = asyncio.create_task(get_all_queue_items(queue))
    await asyncio.gather(*producer_tasks)
    await queue.put(SENTINEL)
    await queue.join()
    items = await collect_queue_items_task
    print(items)
    assert sorted(items) == [1, 2, 3, 4, 5]


asyncio.run(main())

Get all items in an asyncio.Queue and return them

1 Answers