cs.queues 20240318

some Queue subclasses and ducktypes

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

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

Class Channel

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

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

Function 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 polls q.empty().

Function IterablePriorityQueue(capacity=0, name=None)

Factory to create an iterable PriorityQueue.

Function 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.

Method 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]

NullQ = <NullQueue:NullQ blocking=False>

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

Class NullQueue(cs.resources.MultiOpenMixin, cs.context.ContextManagerMixin)

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

Method 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.

Class PushQueue(cs.resources.MultiOpenMixin, cs.context.ContextManagerMixin)

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.

Method PushQueue.__init__(self, name, functor, outQ): 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.

Class QueueIterator(cs.resources.MultiOpenMixin, cs.context.ContextManagerMixin)

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

Class TimerQueue

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

Release Log

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.