Queue¶

An asyncio.Queue equivalent for asyncgui.

Usage¶

import asyncgui as ag
from asyncgui_ext.queue import Queue

async def producer(q):
    for c in "ABC":
        await q.put(c)
        print('produced', c)

async def consumer(q):
    async for c in q:
        print('consumed', c)

q = Queue(capacity=1)
ag.start(producer(q))
ag.start(consumer(q))

produced A
produced B
consumed A
produced C
consumed B
consumed C

API Reference¶

exception Closed¶

Occurs when:

trying to get an item from a queue that is in the CLOSED state.
trying to get an item from an empty queue that is in the HALF_CLOSED state.
trying to put an item into a queue that is in the CLOSED or HALF_CLOSED state.

Order¶

'fifo': First In First Out
'lifo': Last In First Out
'small-first': Smallest One First Out

alias of Literal[‘fifo’, ‘lifo’, ‘small-first’]

class Queue(*, capacity: int | None = None, order: Literal['fifo', 'lifo', 'small-first'] = 'fifo')¶

Parameters:: capacity – Cannot be zero. Unlimited if None.

async __aiter__()¶

Repeats getting an item from the queue until it gets closed.

async for item in queue:
    ...

This is equivalent to:

try:
    while True:
        item = await queue.get()
        ...
except Closed:
    pass

property capacity: int | None¶: Number of items allowed in the queue. None if unlimited.

close()¶: Fully closes the queue. Putting or getting items are no longer allowed, and any items currently held will be discarded.

async get() → Awaitable[Any]¶

item = await queue.get()

get_nowait() → Any¶

item = queue.get_nowait()

half_close()¶: Partially closes the queue. Putting an item is no longer allowed.

property is_empty: bool¶

property is_full: bool¶

property order: Literal['fifo', 'lifo', 'small-first']¶

async put(item) → Awaitable¶

await queue.put(item)

put_nowait(item)¶

queue.put_nowait(item)

property size: int¶: Number of items in the queue. This equals to len(queue).

transfer_items(*_unused)¶

exception QueueException¶: Base class of all the queue-related exceptions.

class QueueState¶

Enum class that represents the state of the Queue.

CLOSED¶

Putting or getting an item is not allowed.

HALF_CLOSED¶

Putting an item is not allowed.

OPENED¶

All operations are allowed.

exception WouldBlock¶: Raised by X_nowait functions if X would block.

Quirk¶

The output of the following code may surprise you.

async def fn1(q, received):
    await q.put('A')
    await q.put('B')
    item = await q.get()
    received.append(item)
    await q.put('C')
    item = await q.get()
    received.append(item)

async def fn2(q, received):
    item = await q.get()
    received.append(item)

received = []
q = Queue(capacity=1, order='fifo')
ag.start(fn1(q, received))
ag.start(fn2(q, received))
print(received)

['B', 'C', 'A']

As you can see, even though fn1 enqueues items in the order A, B, C, the received list ends up with the order B, C, A, which is probably not what you’d expect. In this particular case, you can work around the issue by increasing the queue’s capacity so that fn1 does not block (e.g. to 2). However, to avoid this behavior in all situations, you must rely on a timer to defer the execution of transfer_items(). For example, if you are using Kivy, you want to do:

from asyncgui_ext.queue import Queue
from kivy.clock import Clock

q = Queue(...)
q.transfer_items = Clock.create_trigger(q.transfer_items)

As for tkinter, refer to the example examples/fix_quirk_in_tkinter.py.