cs-queues 20250426
Queue-like items: iterable queues, channels, etc.
Keywords: python2, python3

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

*Latest release 20250426*:
Fix typing annotation, failing on Python 3.8.

Short summary:
* `Channel`: A zero-storage data passage. Unlike a `Queue`, `put(item)` blocks waiting for the matching `get()`.
* `get_batch`: Get up to `max_batch` closely spaced items from the queue `q`. Return the batch.
* `IterablePriorityQueue`: Factory to create an iterable `PriorityQueue`.
* `IterableQueue`: Factory to create an iterable queue. Note that the returned queue is already open and needs a close.
* `ListQueue`: A simple iterable queue based on a `list`.
* `NullQ`: A queue-like object that discards its inputs. Calls to `.get()` raise `queue.Empty`.
* `NullQueue`: A queue-like object that discards its inputs. Calls to `.get()` raise `queue.Empty`.
* `PushQueue`: A puttable object which looks like an iterable `Queue`.
* `QueueIterator`: A `QueueIterator` is a wrapper for a `Queue`like object which presents an iterator interface to collect items. It does not offer the `.get` or `.get_nowait` methods.
* `TimerQueue`: Class to run a lot of "in the future" jobs without using a bazillion Timer threads.

Module contents:
- <a name="Channel"></a>`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.
- <a name="get_batch"></a>`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()`.
- <a name="IterablePriorityQueue"></a>`IterablePriorityQueue(capacity=0, name=None)`: Factory to create an iterable `PriorityQueue`.
- <a name="IterableQueue"></a>`IterableQueue(capacity=0, name=None)`: Factory to create an iterable queue.
  Note that the returned queue is already open
  and needs a close.
- <a name="ListQueue"></a>`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.
- <a name="NullQ"></a>`NullQ = <NullQueue:NullQ blocking=False>`: A queue-like object that discards its inputs.
  Calls to `.get()` raise `queue.Empty`.
- <a name="NullQueue"></a>`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.
- <a name="PushQueue"></a>`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 0x10e368400>)`*:
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.
- <a name="QueueIterator"></a>`Class `QueueIterator(cs.resources.MultiOpenMixin, collections.abc.Iterable, typing.Generic)`: A `QueueIterator` is a wrapper for a `Queue`like 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.
- <a name="TimerQueue"></a>`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 20250426*:
Fix typing annotation, failing on Python 3.8.

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