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 Queuelike 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:
Class Channel: A zero-storage data passage. Unlike a Queue, put(item)blocks waiting for the matchingget()`.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.
ListQueue.__init__(self, queued=None, *, unique=None):
Initialise the queue.
Parameters:
queued is an optional iterable of initial items for the queueunique: optional signature function, default NoneThe 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()raisequeue.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 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.
Class QueueIterator(cs.resources.MultiOpenMixin, collections.abc.Iterable, typing.Generic): A QueueIteratoris a wrapper for aQueuelike object which presents an iterator interface to collect items. It does not offer the .getor.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 20250426: Fix typing annotation, failing on Python 3.8.
Release 20250306:
Release 20250103: PushQueue: mix in RunStateMixin, automatic RunState parameter, honour runstate.cancelled.
Release 20240412:
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:
Release 20230331:
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:
Release 20190812: _QueueIterator: do MultiOpenMixin.init so that str is functional.
Release 20181022: Bugfix Channel, drasticly simplify PushQueue, other minor changes.
Release 20160828:
Release 20150115: More PyPI metadata fixups.
Release 20150111: Initial PyPI release.
FAQs
Queue-like items: iterable queues, channels, etc.
We found that cs-queues demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Application Security
/Security News
Socket CEO Feross Aboukhadijeh and a16z partner Joel de la Garza discuss vibe coding, AI-driven software development, and how the rise of LLMs, despite their risks, still points toward a more secure and innovative future.
Research
/Security News
Threat actors hijacked Toptal’s GitHub org, publishing npm packages with malicious payloads that steal tokens and attempt to wipe victim systems.
Research
/Security News
Socket researchers investigate 4 malicious npm and PyPI packages with 56,000+ downloads that install surveillance malware.