Kodeclik Blog
How to implement priority queues in Python
A priority queue is an “abstract” data structure. Abstract here means we simply define the functionality (ie what it is supposed to do) without explaining how it is implemented.
In a priority queue each element is associated with a priority and the element with the highest (or lowest, depending on how you define it) priority is served before the others.
Unlike a standard first-in-first-out (FIFO) queue, the order in which elements are removed from a priority queue is determined by their priority values rather than their order of insertion.
In many implementations, lower numerical values indicate higher priority, but you can design your own logic depending on the problem you’re solving.
For instance, in the below data structure we have a priority queue of three people and it shows the order in which they "enter" and they order they "leave" the priority queue.
We see that (from the top) we have inserted them in the order of Tom first, then Mary, then Liz. But observe that everybody has a priority value. Thus, Mary has higher priority than Tom who has a higher priority than Liz. As a result when we take elements out of the data structure, we obtain Mary first, then Tom, then Liz even though that was not the order we inserted them.
This is very different from a traditional queue which follows a first-in-first-out (FIFO) semantics or a stack which follows a last-in-first-out (LIFO) semantics. In a priority queue, every element has a priority value which dictates how removals happen.
Below are three different ways to implement a priority queue in Python, each with detailed examples and explanations.
Method 1: Use the built-in heapq module
One of the most common and efficient ways to implement a priority queue in Python is by using the built-in heapq module. The heapq module provides an implementation of the heap queue algorithm, also known as the priority queue algorithm.
In this approach, you represent the queue as a list and use heapq.heappush to add items to the heap and heapq.heappop to remove the smallest item (i.e., the item with the highest priority if you define a smaller number as higher priority).
Consider the below code:
import heapq
def heapq_priority_queue_example():
# Create an empty list to represent our priority queue
pq = []
# Push items onto the priority queue.
# Each item is a tuple where the first element is the priority
# and the second element is the value.
heapq.heappush(pq, (2, "Tom"))
heapq.heappush(pq, (1, "Mary"))
heapq.heappush(pq, (3, "Liz"))
print("Items in the heap (internal state):", pq)
# Pop items off the priority queue.
print("Processing items based on priority:")
while pq:
# heappop returns the smallest element, which is the highest priority.
priority, item = heapq.heappop(pq)
print(f"Priority: {priority}, Item: {item}")
heapq_priority_queue_example()In this example, we create an empty list pq that will serve as our heap-based priority queue. Each time we add an item, we push a tuple where the first element is the priority number. The heap ensures that the item with the smallest priority value (i.e., the highest priority) is always at the front. The while loop pops off items in order of priority until the heap is empty. Notice that the internal state of the heap may not be fully sorted, but the smallest element is always accessible immediately.
The output will be:
Items in the heap (internal state): [(1, 'Mary'), (2, 'Tom'), (3, 'Liz')]
Processing items based on priority:
Priority: 1, Item: Mary
Priority: 2, Item: Tom
Priority: 3, Item: LizIn other words, even though you inserted them in the order of Tom followed by Mary followed by Liz when you pop elements off the queue, you get Mary first, then Tom and then Liz (because of the priority values you specified).
The heap structure guarantees that the smallest element is always at the root, and both insertion and removal operations run efficiently.
Method 2: Use the PriorityQueue class from the built-in queue module
Another approach is to use the PriorityQueue class from the built-in queue module. The interface is very similar to that of a standard queue, but internally it uses a heap to maintain the priority ordering. The usage is straightforward; you simply put items into the queue and get them out.
Here’s how you can use it:
from queue import PriorityQueue
def queue_priority_queue_example():
# Create a PriorityQueue object.
pq = PriorityQueue()
# Put items into the queue. Like with heapq, each item is a tuple
# where the first element is the priority.
pq.put((2, "Tom"))
pq.put((1, "Mary"))
pq.put((3, "Liz"))
print("Processing items based on priority:")
# Retrieve items until the queue is empty.
while not pq.empty():
# get() returns the item with the highest priority (i.e., the smallest value).
priority, item = pq.get()
print(f"Priority: {priority}, Item: {item}")
queue_priority_queue_example()In this code, we instantiate a PriorityQueue object and use its put() method to insert tasks along with their priorities. The internal implementation uses a heap, so the element with the smallest priority value is always retrieved first.
The output will be:
Processing items based on priority:
Priority: 1, Item: Mary
Priority: 2, Item: Tom
Priority: 3, Item: LizMethod 3: Use a sorted list
A third method to implement a priority queue is by using a sorted list. In this approach, each time you insert an item, you place it in the list in sorted order using the bisect module, which allows for binary search insertion. Although this method might be less efficient than a heap for large datasets, it is quite simple and can be suitable for small collections or when the dataset does not change frequently.
Here is an example of a custom priority queue implemented using a sorted list:
import bisect
class SortedPriorityQueue:
def __init__(self):
# Initialize an empty list to store the items.
self._data = []
def push(self, priority, item):
# Use bisect.insort to insert the item in sorted order.
bisect.insort(self._data, (priority, item))
def pop(self):
# Remove and return the first element (i.e., the one with the smallest priority).
if self._data:
return self._data.pop(0)
else:
raise IndexError("pop from an empty priority queue")
def is_empty(self):
return len(self._data) == 0
def sorted_list_priority_queue_example():
# Create an instance of our custom SortedPriorityQueue.
pq = SortedPriorityQueue()
# Push items onto the queue.
pq.push(2, "Tom")
pq.push(1, "Mary")
pq.push(3, "Liz")
print("Processing items based on priority:")
# Pop items until the queue is empty.
while not pq.is_empty():
priority, item = pq.pop()
print(f"Priority: {priority}, Item: {item}")
sorted_list_priority_queue_example()In this implementation, we define a class SortedPriorityQueue that maintains a sorted list _data. When a new item is pushed onto the queue, the bisect.insort() function is used to insert the tuple (priority, item) into the list while keeping the list sorted. The pop() method removes and returns the first element of the list, which corresponds to the element with the smallest priority (i.e., the highest priority in our case).
Although this method has a higher insertion cost compared to heap-based approaches, it is conceptually simple and can be effective when working with a small number of elements.
The output will be, as expected:
Processing items based on priority:
Priority: 1, Item: Mary
Priority: 2, Item: Tom
Priority: 3, Item: LizEach of the above three implementations demonstrates a different way of managing prioritized elements. The heapq module is typically preferred for its efficiency and ease of use in many applications, while queue.PriorityQueue, although we haven’t covered it here, is good for multi-threaded applications. The sorted list approach, although less efficient for large or dynamic datasets, provides an alternative that can be easier to understand and implement for simpler cases.
These examples should give you a solid foundation for implementing and understanding priority queues in Python. Which one is your favorite?
Enjoy this blogpost? Want to learn Python with us? Sign up for 1:1 or small group classes.
