## Remarks

• Priority Queue is a Queue, in which elements will be dequeued based on their priorites(other than timestamp).
• Elements on the queue should be comparable, otherwise we can’t deduce their ralative priorities.
• Priority Queue can be implemented with heap.
• Heap is a tree in which the parent is always larger(max heap) or smaller(min heap) than its children. (Heap property)
• We can implement priority queue using complete binary heap.

operation time
construct O(N)
front O(1)
dequeue O(logN)
enqueue O(logN)

### Implementation

from operator import lt, gt

class HeapQueue:
comparators = {'min': lt, 'max': gt}

def __init__(self, data=None, comparator='min'):
self._heapq = [None] + list(data) if data else [None]
self._comparator = self.comparators.get(comparator, comparator)
self._n = len(self._heapq) - 1
if data: self.heapify()

def _compare(self, i, j):
return self._comparator(self._heapq[i], self._heapq[j])

def __len__(self):
return self._n

def is_empty(self):
return self._n == 0

def _swap(self, i, j):
self._heapq[i], self._heapq[j] = self._heapq[j], self._heapq[i]

def front(self):
assert not self.is_empty(), 'heapq is empty'
return self._heapq[1]

def enqueue(self, x):
self._heapq.append(x)
self._n += 1
self.swim(self._n)

def dequeue(self):
assert not self.is_empty(), 'heapq is empty'
self._swap(1, self._n)
self._n -= 1
self.sink(1)
return self._heapq.pop()

def swim(self, k):
while k > 1 and self._compare(k, k >> 1):
self._swap(k, k >> 1); k >>= 1

def sink(self, k):
n = self._n
while k * 2 <= n:
j = k * 2
if j < n and self._compare(j + 1, j): j += 1
if self._compare(k, j): break
self._swap(k, j); k = j

def heapify(self):
for k in range(self._n >> 1, 0, -1):
self.sink(k)


Simple wrapper over heapq for min priority queue.

from heapq import heapify, heappush, heappop

class PriorityQueue:
def __init__(self, data=[]):
self.data = list(data)
heapify(self.data)

def put(self, x): heappush(self.data, x)

def get(self): return heappop(self.data)

def top(self): return self.data[0]


From python documentation, how to skip the comparison with item

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
priority: int
item: Any = field(compare=False)

class Dummy:
def __init__(self, val): self.val = val

items = [Dummy('b'), Dummy(None), Dummy(Exception('abc'))]
priorities = [1,2,1]
pq = PriorityQueue()

for p, i in zip(priorities, items):
pq.put(PrioritizedItem(priority=p, item=i))

for _ in range(3):
print(pq.get().item.val)

# gets b, abc, None


pq = PriorityQueue()

for o, (p, i) in enumerate(zip(priorities, items)):
pq.put(PrioritizedItem(priority=(p, -o), item=i))

for _ in range(3):
print(pq.get().item.val)

# gets abc, b, None


Hack the prioritized item to be used as max priority queue

@dataclass(order=False)
class PrioritizedItem:
priority: int
item: Any = field(compare=False)

def __lt__(self, other): return self.priority > other.priority

pq = PriorityQueue()

for o, (p, i) in enumerate(zip(priorities, items)):
pq.put(PrioritizedItem(priority=(p, -o), item=i))

for _ in range(3):
print(pq.get().item.val)

# gets None, b, abc

pq = PriorityQueue()

for o, (p, i) in enumerate(zip(priorities, items)):
pq.put(PrioritizedItem(priority=(p, o), item=i))

for _ in range(3):
print(pq.get().item.val)

# gets None, abc, b