Queutils [Queue Utils] is a package of handy Python queue classes:
- AsyncQueue - An
asyncwrapper for non-asyncqueue.Queue - IterableQueue - An
AsyncIterablequeue that terminates when finished - EventCounterQueue - An
IterableQueuefor counting events inasyncthreads - FileQueue - Builds an
IterableQueue[pathlib.Path]of filenames from files/dirs given as input
AsyncQueue is a async wrapper for non-async queue.Queue. It can be used to create
an asyncio.Queue compatible interface to a (non-async) managed multiprocessing.Queue and thus enable async code in parent/child processes to communicate over multiprocessing.Queue as it were an asyncio.Queue. Uses sleep() for get()/put() if the queue is empty/full.
asyncio.Queuecompatiblequeue.Queuesupportmultiprocessing.Queuesupport
IterableQueue is an asyncio.Queue subclass that is AsyncIterable[T] i.e. it can be
iterated in async for loop. IterableQueue terminates automatically when the queue has been filled and emptied.
The IterableQueue requires "producers" (functions adding items to the queue) to register themselves with add_producer() call. It keeps count of registered producers. When a producer "finishes" adding items to the queue,
it needs to unregister itself with finish_producer() call. Once all the registered
producers are "finished", the queue enters into "filled" state and no new items can be added. Once a "filled" queue has been emptied, the queue becomes "done" and
all new get() calls to the queue will raise QueueDone exception.
asyncio.Queueinterface,_nowait()methods are experimentalAsyncIterablesupport:async for item in queue:- Automatic termination of the consumers with
QueueDoneexception when the queue has been emptied - Producers must be registered with
add_producer()and they must notify the queue withfinish_producer()once they have finished adding items - Countable interface to count number of items task_done() through
countproperty
EventCounterQueue can be used to count named events (default event is count) between async threads. async worker threads call queue.send(event="event_name", N=amount). The receving end can either receive() a single event or listen() all events and return collections.defaultdict[str, int] as a result.
- Supports multiple producers and a single listener
- Default event is
count
FileQueue builds a queue (IterableQueue[pathlib.Path]) of the matching
files found based on search parameters given. It can search both list of files or directories or
mixed. Async method FileQueue.mk_queue() searches subdirectories of given directories.
- Input can be given both as
strandpathlib.Path exclude: boolexclusive or inclusive filtering. Default isFalse.case_sensitive: boolcase sensitive filtering (use offnmatchorfnmatchcase). Default isTrue.follow_symlinks: boolwhether to follow symlinks. Default isFalse.