cs-queues 20250306

Queue-like items: iterable queues, channels, etc.

Queue-like items: iterable queues, channels, etc.

Latest release 20250306:

• QueueIterator: annotate as Iterable[Any].
• Drop obsolete pretence of python 2 support, embrace the f-string.
• Some other minor updates.

Class Channel

A zero-storage data passage. Unlike a Queue, put(item) blocks waiting for the matching get().

Channel.__init__(self): pylint: disable=consider-using-with

Channel.__call__(self, *a): Call the Channel. With no arguments, do a .get(). With an argument, do a .put().

Channel.__iter__(self): A Channel is iterable.

Channel.__next__(self): next(Channel) calls Channel.get().

Channel.__str__(self): pylint: disable=consider-using-with

Channel.close(self): Close the Channel, preventing further put()s.

Channel.get(self): Read a value from the Channel. Blocks until someone put()s to the Channel.

Channel.put(self, value): Write a value to the Channel. Blocks until a corresponding get() occurs.

get_batch(q, max_batch=128, *, poll_delay=0.01)

Get up to max_batch closely spaced items from the queue q. Return the batch. Raise queue.Empty if the first q.get() raises.

Block until the first item arrives. While the batch's size is less that max_batch and there is another item available within poll_delay seconds, append that item to the batch.

This requires get_batch() to be the sole consumer of q for correct operation as it makes decisions based on q.empty().

IterablePriorityQueue(capacity=0, name=None)

Factory to create an iterable PriorityQueue.

IterableQueue(capacity=0, name=None)

Factory to create an iterable queue. Note that the returned queue is already open and needs a close.

Class ListQueue

A simple iterable queue based on a list.

ListQueue.__init__(self, queued=None, *, unique=None): Initialise the queue.

Parameters:

• queued is an optional iterable of initial items for the queue
• unique: optional signature function, default None

The unique parameter provides iteration via the cs.seq.unrepeated iterator filter which yields only items not seen earlier in the iteration. If unique is None or False iteration iterates over the queue items directly. If unique is True, iteration uses the default mode where items are compared for equality. Otherwise unique may be a callable which produces a value to use to detect repetitions, used as the cs.seq.unrepeated signature parameter.

Example:

>>> items = [1, 2, 3, 1, 2, 5]
>>> list(ListQueue(items))
[1, 2, 3, 1, 2, 5]
>>> list(ListQueue(items, unique=True))
[1, 2, 3, 5]

ListQueue.__bool__(self): A ListQueue looks a bit like a container, and is false when empty.

ListQueue.__iter__(self): A ListQueue is iterable.

ListQueue.__next__(self): Iteration gets from the queue.

ListQueue.append(self, item): Append an item to the queue, aka put.

ListQueue.extend(self, items): Convenient/performant queue-lots-of-items.

ListQueue.get(self): Get pops from the start of the list.

ListQueue.insert(self, index, item): Insert item at index in the queue.

ListQueue.prepend(self, items, offset=0): Insert items at offset (default 0, the front of the queue).

ListQueue.put(self, item): Put appends to the queue.

NullQ = <NullQueue:NullQ blocking=False>

A queue-like object that discards its inputs. Calls to .get() raise queue.Empty.

Class NullQueue(cs.resources.MultiOpenMixin)

A queue-like object that discards its inputs. Calls to .get() raise queue.Empty.

NullQueue.__init__(self, blocking=False, name=None): Initialise the NullQueue.

Parameters:

• blocking: optional; if true, calls to .get() block until .shutdown(); default: False.
• name: optional name for this NullQueue.

NullQueue.get(self): Get the next value. Always raises queue.Empty. If .blocking, delay until .shutdown().

NullQueue.put(self, item): Put a value onto the queue; it is discarded.

NullQueue.shutdown(self): Shut down the queue.

NullQueue.startup(self): Start the queue.

Class PushQueue(cs.resources.MultiOpenMixin, cs.resources.RunStateMixin)

A puttable object which looks like an iterable Queue.

In this base class, calling .put(item) calls functor supplied at initialisation to trigger a function on data arrival whose iterable of results are put onto the output queue.

As an example, the cs.pipeline.Pipeline class uses subclasses of PushQueue for each pipeline stage, overriding the .put(item) method to mediate the call of functor through cs.later.Later as resource controlled concurrency.

PushQueue.__init__(self, name: str, functor: Callable[[Any], Iterable], outQ, runstate: Optional[cs.resources.RunState] = <function uses_runstate.<locals>.<lambda> at 0x10ff9a2a0>): Initialise the PushQueue with the callable functor and the output queue outQ.

Parameters:

• functor is a one-to-many function which accepts a single item of input and returns an iterable of outputs; it may be a generator. These outputs are passed to outQ.put individually as received.
• outQ is a MultiOpenMixin which accepts via its .put() method.

PushQueue.put(self, item): Receive a new item, put the results of functor(item) onto self.outQ.

Subclasses might override this method, for example to process the result of functor differently, or to queue the call to functor(item) via some taks system.

PushQueue.startup_shutdown(self): Open/close the output queue.

Class QueueIterator(cs.resources.MultiOpenMixin, collections.abc.Iterable, typing.Generic)

A QueueIterator is a wrapper for a Queuelike object which presents an iterator interface to collect items. It does not offer the .get or .get_nowait methods.

QueueIterator.__iter__(self): Iterable interface for the queue.

QueueIterator.__next__(self): Return the next item from the queue. If the queue is closed, raise StopIteration.

QueueIterator.empty(self): Test if the queue is empty.

QueueIterator.iter_batch(self, batch_size=1024): A generator which yields batches of items from the queue. The default batch_size is 1024.

QueueIterator.join(self): Wait for the queue items to complete.

QueueIterator.next(self): Return the next item from the queue. If the queue is closed, raise StopIteration.

QueueIterator.next_batch(self, batch_size=1024, block_once=False): Obtain a batch of immediately available items from the queue. Up to batch_size items will be obtained, default 1024. Return a list of the items. If the queue is empty an empty list is returned. If the queue is not empty, continue collecting items until the queue is empty or the batch size is reached. If block_once is true, wait for the first item; this mode never returns an empty list except at the end of the iterator.

QueueIterator.put(self, item, *args, **kw): Put item onto the queue. Warn if the queue is closed. Raises ValueError if item is the sentinel.

QueueIterator.startup_shutdown(self): MultiOpenMixin support; puts the sentinel onto the underlying queue on the final close.

QueueIterator.task_done(self): Report that an item has been processed.

Class TimerQueue

Class to run a lot of "in the future" jobs without using a bazillion Timer threads.

TimerQueue.add(self, when, func): Queue a new job to be called at 'when'. 'func' is the job function, typically made with functools.partial.

TimerQueue.close(self, cancel=False): Close the TimerQueue. This forbids further job submissions. If cancel is supplied and true, cancel all pending jobs. Note: it is still necessary to call TimerQueue.join() to wait for all pending jobs.

TimerQueue.join(self): Wait for the main loop thread to finish.

Release Log

Release 20250306:

• QueueIterator: annotate as Iterable[Any].
• Drop obsolete pretence of python 2 support, embrace the f-string.
• Some other minor updates.

Release 20250103: PushQueue: mix in RunStateMixin, automatic RunState parameter, honour runstate.cancelled.

Release 20240412:

• QueueIterator: new next_batch and iter_batch methods for collecting items in batches.
• QueueIterator: bugfix put(), do not hold the lock around the internal Queue.put, can deadlock the corresponding next/get.

Release 20240318: QueueIterator.put: be more rigorous with the put + item count increment.

Release 20240316: Fixed release upload artifacts.

Release 20240305: New get_batch(q) to fetch closely spaced batches from a Queue.

Release 20240211: ListQueue: new unique parameter which uses cs.seq.unrepeated in iteration.

Release 20231129:

• ListQueue: add len, str, repr.
• QueueIterator: use a mutex around the item counting.

Release 20230331:

• QueueIterator.startup_shutdown: try/finally for the terminating _put(self.sentinel).
• QueueIterator.str: show the underlying queue.
• QueueIterator: drop call to self.finalise(), no longer available or needed.

Release 20221228: Minor doc update.

Release 20221207: PushQueue: modernise the MutiOpenMixin startup_shutdown, do both open and close of self.outQ.

Release 20220918: Expose formerly private _QueueIterator as QueueIterator, use a per-queue sentinel from cs.obj.Sentinel.

Release 20220805: ListQueue: add "append" synonym for "put" in keeping with the list-ish flavour.

Release 20220605: ListQueue: extend/prepend: reject str explicitly - although iterable, it is almost never what is intended.

Release 20220317: Add missed import.

Release 20220313: New ListQueue.prepend(items[,offset=0]) method.

Release 20211116: ListQueue: new insert() method.

Release 20210924: Channel: make a Channel iterable.

Release 20210913: New ListQueue simple iterable queue based on a list with list-like .append and .extend.

Release 20201025: Drop obsolete call to MultiOpenMixin.init.

Release 20200718: _QueueIterator: set finalise_later via new MultiOpenMixin property, required by recent MultiOpenMixin change.

Release 20200521: IterableQueue,IterablePriorityQueue: simplify wrappers, bypasses weird bug from overengineering these.

Release 20191007:

• PushQueue: improve str.
• Clean lint, drop cs.obj dependency.

Release 20190812: _QueueIterator: do MultiOpenMixin.init so that str is functional.

Release 20181022: Bugfix Channel, drasticly simplify PushQueue, other minor changes.

Release 20160828:

• Use "install_requires" instead of "requires" in DISTINFO.
• TimerQueue.add: support optional *a and **kw arguments for func.
• Many bugfixes and internal changes.

Release 20150115: More PyPI metadata fixups.

Release 20150111: Initial PyPI release.