dataclasses is just a convenience method to avoid having to create a lot of boilerplate code.
You don't actually have to create a class. A tuple with a unique counter value too:
from itertools import count
unique = count()
q.put((priority, next(unique), item))
so that ties between equal priority are broken by the integer that follows; because it is always unique the item value is never consulted.
You can also create a class using straight-up rich comparison methods, made simpler with @functools.total_ordering:
from functools import total_ordering
@total_ordering
class PrioritizedItem:
def __init__(self, priority, item):
self.priority = priority
self.item = item
def __eq__(self, other):
if not isinstance(other, __class__):
return NotImplemented
return self.priority == other.priority
def __lt__(self, other):
if not isinstance(other, __class__):
return NotImplemented
return self.priority < other.priority
python - Using queue.PriorityQueue, not caring about comparisons - Stack Overflow
python - How to iterate over a Priority Queue? - Stack Overflow
python - How to put items into priority queues? - Stack Overflow
Is there a better priority queue?
Videos
dataclasses is just a convenience method to avoid having to create a lot of boilerplate code.
You don't actually have to create a class. A tuple with a unique counter value too:
from itertools import count
unique = count()
q.put((priority, next(unique), item))
so that ties between equal priority are broken by the integer that follows; because it is always unique the item value is never consulted.
You can also create a class using straight-up rich comparison methods, made simpler with @functools.total_ordering:
from functools import total_ordering
@total_ordering
class PrioritizedItem:
def __init__(self, priority, item):
self.priority = priority
self.item = item
def __eq__(self, other):
if not isinstance(other, __class__):
return NotImplemented
return self.priority == other.priority
def __lt__(self, other):
if not isinstance(other, __class__):
return NotImplemented
return self.priority < other.priority
See priority queue implementation notes - just before the section you quoted (regarding using dataclasses) it tells you how to do it whitout them:
... is to store entries as 3-element list including the priority, an entry count, and the task. The entry count serves as a tie-breaker so that two tasks with the same priority are returned in the order they were added. And since no two entry counts are the same, the tuple comparison will never attempt to directly compare two tasks.
So simply add your items as 3rd element in a tuple (Prio, Count, YourElem) when adding to your queue.
Contreived example:
from queue import PriorityQueue
class CompareError(ValueError): pass
class O:
def __init__(self,n):
self.n = n
def __lq__(self):
raise CompareError
def __repr__(self): return str(self)
def __str__(self): return self.n
def add(prioqueue,prio,item):
"""Adds the 'item' with 'prio' to the 'priorqueue' adding a unique value that
is stored as member of this method 'add.n' which is incremented on each usage."""
prioqueue.put( (prio, add.n, item))
add.n += 1
# no len() on PrioQueue - we ensure our unique integer via method-param
# if you forget to declare this, you get an AttributeError
add.n = 0
h = PriorityQueue()
add(h, 7, O('release product'))
add(h, 1, O('write spec 3'))
add(h, 1, O('write spec 2'))
add(h, 1, O('write spec 1'))
add(h, 3, O('create tests'))
for _ in range(4):
item = h.get()
print(item)
Using h.put( (1, O('write spec 1')) ) leads to
TypeError: '<' not supported between instances of 'O' and 'int'`
Using def add(prioqueue,prio,item): pushes triplets as items wich have guaranteed distinct 2nd values so our O()-instances are never used as tie-breaker.
Output:
(1, 2, write spec 3)
(1, 3, write spec 2)
(1, 4, write spec 1)
(3, 5, create tests)
see MartijnPieters answer @here for a nicer unique 2nd element.
queue.PriorityQueue is actually implemented using a list, and the put/get methods use heapq.heappush/heapq.heappop to maintain the priority ordering inside that list. So, if you wanted to, you could just iterate over the internal list, which is contained in the queue attribute:
>>> from queue import PriorityQueue
>>> q = PriorityQueue()
>>> q.put((5, "a"))
>>> q.put((3, "b"))
>>> q.put((25, "c"))
>>> q.put((2, "d"))
>>> print(q.queue)
[(2, 'd'), (3, 'b'), (25, 'c'), (5, 'a')]
The PriorityQueue is implemented as binary heap, which is implemented using a list (array) in python. To iterate over the queue you need to know the rules about where children are stored in the list.
The rules being that all nodes have two children, unless they are the last node to have children in which case they may have one child instead. All nodes that appear after the last node to have children will have zero children (duh).
A node's children are stored in relation to the node's own position in the list. Where i is the index of the nodes in the list then it's children are stored at:
2 * i + 12 * i + 2
However, the only requirement of a heap is that all a node's children have a value greater than or equal to the node's value (or greater than depending on the implementation).
For instance, in the above linked wiki page about binrary heap's you'll find the following image. The first item in the queue is the root. Quite obvious. The second item is the larger of the root's children. However, the third item could either be the remaining node of the root node, or either of the children of the second node in the queue. That is, the third item in the queue (25) could have been in the same position as either 19 or 1.
Thus, to iterate over the queue you need to keep track of all the currently "viewable" nodes. For instance:
def iter_priority_queue(queue):
if queue.empty():
return
q = queue.queue
next_indices = [0]
while next_indices:
min_index = min(next_indices, key=q.__getitem__)
yield q[min_index]
next_indices.remove(mix_index)
if 2 * min_index + 1 < len(q):
next_indices.append(2 * min_index + 1)
if 2 * min_index + 2 < len(q):
next_indices.append(2 * min_index + 2)
The method can be monkey patched onto queue.PriorityQueue if you're feeling lazy, but I would encourage you to implement your own priority queue class using the heapq module as the PriorityQueue comes with a lot of excess functionality (mainly it is thread safe which almost certainly don't need). It should be noted that the above method is not thread safe. If another thread modifies the queue whilst it is being iterated over then the above method will start yielding the wrong numbers and if you're lucky it may produce an exception.
Just use the second item of the tuple as a secondary priority if a alphanumeric sort on your string data isn't appropriate. A date/time priority would give you a priority queue that falls back to a FIFIO queue when you have multiple items with the same priority. Here's some example code with just a secondary numeric priority. Using a datetime value in the second position is a pretty trivial change, but feel free to poke me in comments if you're not able to get it working.
Code
import Queue as queue
prio_queue = queue.PriorityQueue()
prio_queue.put((2, 8, 'super blah'))
prio_queue.put((1, 4, 'Some thing'))
prio_queue.put((1, 3, 'This thing would come after Some Thing if we sorted by this text entry'))
prio_queue.put((5, 1, 'blah'))
while not prio_queue.empty():
item = prio_queue.get()
print('%s.%s - %s' % item)
Output
1.3 - This thing would come after Some Thing if we didn't add a secondary priority
1.4 - Some thing
2.8 - super blah
5.1 - blah
Edit
Here's what it looks like if you use a timestamp to fake FIFO as a secondary priority using a date. I say fake because it's only approximately FIFO as entries that are added very close in time to one another may not come out exactly FIFO. I added a short sleep so this simple example works out in a reasonable way. Hopefully this helps as another example of how you might get the ordering you're after.
import Queue as queue
import time
prio_queue = queue.PriorityQueue()
prio_queue.put((2, time.time(), 'super blah'))
time.sleep(0.1)
prio_queue.put((1, time.time(), 'This thing would come after Some Thing if we sorted by this text entry'))
time.sleep(0.1)
prio_queue.put((1, time.time(), 'Some thing'))
time.sleep(0.1)
prio_queue.put((5, time.time(), 'blah'))
while not prio_queue.empty():
item = prio_queue.get()
print('%s.%s - %s' % item)
As far as I know, what you're looking for isn't available out of the box. Anyway, note that it wouldn't be hard to implement:
from Queue import PriorityQueue
class MyPriorityQueue(PriorityQueue):
def __init__(self):
PriorityQueue.__init__(self)
self.counter = 0
def put(self, item, priority):
PriorityQueue.put(self, (priority, self.counter, item))
self.counter += 1
def get(self, *args, **kwargs):
_, _, item = PriorityQueue.get(self, *args, **kwargs)
return item
queue = MyPriorityQueue()
queue.put('item2', 1)
queue.put('item1', 1)
print queue.get()
print queue.get()
Example output:
item2
item1
I know python has heapq and queue.priorityqueue but honestly, both of them are really cumbersome compared to Java's priorityqueue. I find it tedious to have to insert a tuple, with the first element in the tuple defining the priority. Also, it makes it hard to write more complex comparisons. Is there a way we can pass in a comparator to the Priorityqueue? I know it's possible to define classes with their own comparator method, but again, this is really tedious and I'm looking for something as close as possible to Java's PQ.