Async

Helper module to add some utils to Python’s asyncio.

class cli2.asyncio.Queue(*args, num_workers=12, **kwargs)

An async queue with worker pool for concurrent task processing.

Extends asyncio.Queue to manage a pool of worker tasks that process items from the queue concurrently.

# will run 2 at the time
queue = cli2.Queue(num_workers=2)
# call like asyncio.run
await queue.run(foo(), bar(), other())

num_workers

Number of concurrent workers (default: 12)

results

List of results from completed tasks, order of results not garanteed due to concurrency.

async run(*tasks)

Run tasks through the worker pool.

Parameters:

tasks – Coroutines

async worker()

Worker task that processes items from the queue.

Continuously gets tasks from the queue, executes them, and stores results. Handles exceptions by logging them.

cli2.asyncio.async_iter(obj)

Check if an object is an async iterable.

async cli2.asyncio.async_resolve(result, output=False)

Recursively resolve awaitables and async iterables.

Parameters:

• result – The awaitable or async iterable to resolve

• output – If True, print results as they are resolved. If False, collect results.

Returns:

The resolved value(s). If output is True, returns None. If output is False, returns a list of resolved values from async iterables.

cli2.asyncio.async_run(coroutine)

Run an async coroutine in the current event loop or create a new one.

If an event loop is already running, creates a task in that loop. If no event loop is running, creates a new one and runs the coroutine.

Parameters:

coroutine – The coroutine to run (return value of an async function)

Returns:

The result of the coroutine execution