semvec/pypi-packages huggingface dataset:
aioqueueext 0.2.1
Asyncio queues with extended functionality: peeking, setting put and get callbacks.
Requires Python: <4,>=3.7
Keywords: asyncio,queue

# aioqueueext

A package that provides asyncio Queues with additional functionality.

## Work-in-Progress

The repository contains modules extracted from my other project and was refactored as a separate package.

In the current version, I have not verified all of the functions.

Additional functions I plan to implement are:
- `return_when_*()` - async functions to ease synchronization tasks
- `set_on_get_callback()`
- `set_on_put_callback()`
- `peek_nowait()` - returns the "up-next" item without removing it from the queue
- `peek_and_get()` - async peek and conditionally get (pop) an item from the queue

## Examples

### Sync Peeking

```
async def sync_peeking_example() -> None:
    queue1 = QueueExt()

    await queue1.put("apple")

    item = queue1.peek_nowait()
    print(f"first peek: {item}")  # "apple"

    await queue1.put("banana")

    item = queue1.peek_nowait()
    print(f"second peek: {item}")  # "apple"

    item = await queue1.get()  # popped "apple"

    item = queue1.peek_nowait()
    print(f"third peek: {item}")  # "banana"

    item = await queue1.get()  # popped "banana"

    # the next peek raises the QueueEmpty exception
    try:
        print(f"fourth peek: {queue1.peek_nowait()}")
    except asyncio.QueueEmpty:
        print("fourth peek failed: QueueEmpty")

```