Python Priority Queue using queue, heapq and bisect Modules

The priority queue is an abstract data type that is like a regular queue, but each element in the queue has a “priority” associated with it. In a priority queue, an element with high priority is served before an element with low priority. If two elements have the same priority, they are served according to their order in the priority queue.

The main difference between a regular queue and a priority queue is that a regular queue serves the elements in FIFO order whereas as a priority queue elements are served based on priority. The priority queues are used in several usecases, such as job scheduling algorithms and message processing systems.

1. Importing PriorityQueue using ‘queue‘ Module

The queue module has an inbuilt class PriorityQueue. This priority queue can accept comparable items and the lowest-valued entries are retrieved first. It provides the best and worst case performance with time complexity of O(log n).

If the queue elements are not comparable, the data can be wrapped in a class (such as tuples) that ignores the data item and only compares the priority number:

from dataclasses import dataclass, field
from typing import Any

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

Now we can use the queue.put() and queue.get() methods to push and pop elements from the queue.

from queue import PriorityQueue

q = PriorityQueue()

q.put(Data(5, "how"))
q.put(Data(4, "to"))
q.put(Data(1, "do"))
q.put(Data(3, "in"))
q.put(Data(2, "java"))

for i in range(5):
    print(q.get())

The program output.

Data(priority=1, item='do')
Data(priority=2, item='java')
Data(priority=3, item='in')
Data(priority=4, item='to')
Data(priority=5, item='how')

2. Implementing Priority Queue using heapq Module

2.1. Implementation

The heapq module provides an implementation of the heap queue algorithm, also known as the priority queue algorithm. Heaps are binary trees for which every parent node has a value less than or equal to any of its children, the smallest element is always the root, heap[0].

We use the following methods to push and pop the queue elements:

• heappush(): pushes the value item onto the heap, maintaining the heap invariant.
• heappop(): pops and returns the smallest item from the heap, maintaining the heap invariant.

The following Python program uses the heapq module to implement a simple priority queue:

import heapq

class PriorityQueue:
    def __init__(self):
        self._queue = []
        self._index = 0

    def push(self, item, priority):
        heapq.heappush(self._queue, (-priority, self._index, item))
        self._index += 1

    def pop(self):
        return heapq.heappop(self._queue)[-1]

2.2. Demo

Let’s see an example of how to use the above-created priority queue.

class Item:
  def __init__(self, name):
    self.name = name
  def __repr__(self):
    return 'Item({!r})'.format(self.name)
 
>>> q = PriorityQueue()
 
>>> q.push(Item('how'), 1)
>>> q.push(Item('to'), 5)
>>> q.push(Item('do'), 4)
>>> q.push(Item('in'), 2)
>>> q.push(Item('java'), 1)
 
>>> q.pop()
Item('to')    #5
 
>>> q.pop()
Item('do')    #4
 
>>> q.pop()
Item('in')    #2
 
>>> q.pop()
Item('how')   #1
 
>>> q.pop()
Item('java')  #1

3. Implementing Priority Queue using bisect Module

3.1. Implementation

The bisect module, from the standard Python library, is very handy for maintaining a sorted list without having to sort the list after each insertion. The module is called bisect because it uses a basic bisection algorithm to do its work.

Its insort() method is called with a first argument that is a currently sorted list and an arbitrary second argument. The function inserts the second argument in the list so that the list remains sorted in logarithmic (O(log(N))) time.

Here, we insert the pair (priority, data). Since pairs are compared lexicographically, this means that data will be placed in increasing order of priority.

import bisect

class PriorityQueue:
    def __init__(self):
        self.queue = []

    def insert(self, data, priority):
        bisect.insort(self.queue, (priority, data))

    def pop(self):
        return self.queue.pop()[1]

3.2. Demo

Let’s see an example of how to use the above-created priority queue.

q = PriorityQueue()

q.insert('how',5)
q.insert('to',4)
q.insert('do',5)
q.insert('in',8)
q.insert('java',1)

for i in range(5):
    print(q.pop())

# Prints
in
how
do
to
java

4. Conclusion

In this simple python tutorial, we learned to implement a priority queue in Python using the queue.PriorityQueue class, heapq and bisect modules with examples.

Happy Learning !!

Source Code on Github