In one of the tutorials, we have discussed queues in Python. There is a special kind of queue called a priority queue with a different set of rules for removing items. In this tutorial, we will talk about the Python priority queue.
Specifically, we will discuss how to implement a priority queue from scratch using a list as the underlying data structure. Also, we will discuss how to make use of a Python priority queue.
Introduction to Priority Queue in Python Programming
A Queue is a sequence of objects processed in the order in which they arrive. That is, the first item to arrive will be the first item to be processed, and then removed from the queue.
What is a Priority Queue in Python?
If we add a restriction to a queue such that items are removed based on their priority, we have a Priority Queue.
In a priority queue, two options exist to dequeue a prioritized item:
- If we dequeued items based on minimum priority, we have a min-priority queue.
- On the other hand, if we decided to dequeue items with maximum priority, such a priority queue is called a max-priority queue.
Priority Queue Operations
A priority queue data structure has the following operation:
- Add an item to the priority queue, or Enqueue an item.
- Remove an item from the priority queue, or Dequeue an item. If the queue is a min-priority queue, the queue removes and returns the item with the least priority.
- Find out if a queue is empty.
- Check if a queue is full if the queue has a fixed capacity
- Return the front item of a queue. If the queue is a min-priority queue, we will get the item with the least priority.
- Get the rear item of a queue. If the queue is a min-priority queue, the item returned will have the highest priority.
- Print out the whole queue
The following section will show a possible implementation of a priority queue by making use of a list data structure.
How to Implement Priority Queue Python using a list?
We can implement a priority queue by using a list. Our implementation will be a priority queue whose capacity can be fixed by the user.
In addition to that, our priority queue should also allow us to specify whether our queue is a min-priority or max-priority queue.
The following code shows the complete implementation. In this, we have made use of python bisect.
import bisect as bi
class PriorityQueue:
def __init__(self, capacity=0, max_priority=False):
super().__init__()
self.__items = list()
self.__capacity = 0 if capacity < 0 else capacity
self.__max_priority = max_priority
def is_empty(self):
return (len(self.__items) == 0)
def is_full(self):
if (self.__capacity == 0):
# this is an infinite capacity queue
# can never be filled
return False
else:
# this is a fixed capacity queue
return (len(self.__items) == self.__capacity)
def head(self):
if not(self.is_empty()):
if not self.__max_priority:
# get the first item of the list
return self.__items[0]
else:
# get the last item of the list
return self.__items[-1]
else:
# return None since queue is empty
return None
def tail(self):
if not(self.is_empty()):
if not self.__max_priority:
# get the last item of the list
return self.__items[-1]
else:
# get the first item of the list
return self.__items[0]
else:
# return None since queue is empty
return None
def get_capacity(self):
# return the capacity of the queue
return self.__capacity
def __len__(self):
return len(self.__items)
def __str__(self):
if not self.__max_priority:
# string representation for min priority queue
return f"priority_queue({str(self.__items)})"
else:
# string representation for max priority queue
return f"priority_queue({str(self.__items[::-1])})"
def enqueue(self, item, priority):
if not(self.is_full()):
# use bisect_right to get the insertion point in the list
# so as to keep the list sorted
idx = bi.bisect_right(self.__items, (priority, item))
# insert the new item
self.__items.insert(idx, (priority, item))
# enqueue succeeded
return True
else:
# enqueue failed since queue is full
return False
def dequeue(self):
if not(self.is_empty()):
if not self.__max_priority:
# use the parent class implementation
return self.dequeue()
else:
# pop the last item of the list
return self.__items.pop()
else:
# return None since queue is empty
return None
With our PriorityQueue class in place, we can demonstrate how to make use of it using some code.
Example
The next code defines a status function, display_status(), that takes a queue as input, and prints the state of the queue to the console.
def display_status(q):
print("Queue Status__________________________________________")
print(q)
print(f"Front: {q.head()}")
print(f"Rear: {q.tail()}")
print(f"Length: {len(q)}")
capacity_as_string = f"Finite Capacity => {self.__capacity}" if (self.__capacity > 0) else "Infinite Capacity"
print(f"Capacity: {capacity_as_string}")
print(f"Is Empty: {q.is_empty()}")
print(f"Is Full: {q.is_full()}")
print("______________________________________________________")
print("\n")
The next code segment creates a max-priority queue and calls display_status() to display the status of the new, empty priority queue object.
# create a max priority
patient_queue = PriorityQueue(max_priority=True)
# display status of priority queue
display_status(patient_queue)
Output:
Queue Status__________________________________________
priority_queue([])
Front: None
Rear: None
Length: 0
Capacity: Infinite Capacity
Is Empty: True
Is Full: False
______________________________________________________The output clearly shows that our initial queue is empty. The capacity of our queue is “infinite” since we did not specify any capacity for it.
Next, let us add some patients to the queue. Each patient has a priority based on how serious the patient’s condition is. A priority of 1 is the lowest, meaning the condition is “Least Critical”. A condition of 2 is “Critical”, while a condition of 3 is “Very Critical”.
# add 7 patients to the queue
# we have three levels of priority:
# least critical = 1
# critical = 2
# very critical = 3
patients_with_priority = [('Gary', 2), ('Priyanka', 1), ('Chang', 2), ('Faulkner', 3),
('Ahmad', 1), ('Ming', 2), ('Carlos', 3)]
for name, priority in patients_with_priority:
print(f"Adding patient {name}")
patient_queue.enqueue(name, priority)
print("\n")
The list shows the order in which the patients join the queue. We can see this from the output.
Output:
Adding patient Gary
Adding patient Priyanka
Adding patient Chang
Adding patient Faulkner
Adding patient Ahmad
Adding patient Ming
Adding patient CarlosHowever, when we run the following code
# display status of priority queue
display_status(patient_queue)
We get the status message
Queue Status__________________________________________
priority_queue([(3, 'Faulkner'), (3, 'Carlos'), (2, 'Ming'), (2, 'Gary'), (2, 'Chang'), (1, 'Priyanka'), (1, 'Ahmad')])
Front: (3, 'Faulkner')
Rear: (1, 'Ahmad')
Length: 7
Capacity: Infinite Capacity
Is Empty: False
Is Full: False
______________________________________________________From this status display, we can see that the priority queue has re-arranged the patients in order of priority; from high priority to low priority.
So, we have the most critical patients in front and the least critical patients at the rear of the queue.
Finally, we want to treat the patients and remove them from the queue by some calls to dequeue() method.
# remove all patients from the queue
for count in range(0, len(patient_queue)):
print(f"Treating patient {patient_queue.head()[1]} ...")
print(f"{patient_queue.dequeue()[1]} left queue\n")
Output:
Treating patient Faulkner ...
Faulkner left queue
Treating patient Carlos ...
Carlos left queue
Treating patient Ming ...
Ming left queue
Treating patient Gary ...
Gary left queue
Treating patient Chang ...
Chang left queue
Treating patient Priyanka ...
Priyanka left queue
Treating patient Ahmad ...
Ahmad left queueYou will notice that, unlike in a normal queue, the patients leave the queue based on how serious their cases were. On the other hand, patients in a normal queue would have left based on who arrived first.
We can confirm that the loop in the previous code has indeed emptied our queue by running the status display code.
# display status of priority queue
display_status(patient_queue)
Output:
Queue Status__________________________________________
priority_queue([])
Front: None
Rear: None
Length: 0
Capacity: Infinite Capacity
Is Empty: True
Is Full: False
______________________________________________________Problem with list implementation
The operation that searches for where to insert is fast (it just takes O(log(n)) time). However, the actual insertion operation is a slow O(n) operation. Therefore, we can make use of a list to implement a queue if our queue contains only a few items. If we need to handle more items, we have a better option.
queue.PriorityQueue
Python has a PriorityQueue class implementation which can be found in python’s queue module.
PriorityQueue inherits from class Queue in the same module. As a result, it has all the features of the Queue class. However, it dequeues items based on priority.
A python priority queue object has the following important methods:
put(): Enqueues a new itemget(): Dequeues the front itemempty(): Checks if the priority queue is emptyfull(): Checks if the priority queue is full
However, python’s PriorityQueue is a min-priority queue because it dequeues the item having the least priority.
The following example illustrates this.
from queue import PriorityQueue
pq = PriorityQueue()
items = [(3, 'Maureen'), (1, 'Greg'), (3, 'Mukhtar'), (2, 'Meera')]
# add the items to the priority queue
for item in items:
pq.put(item)
# remove the items from queue to demonstrate that python's
# priority queue is a min-priority queue
while not pq.empty():
item = pq.get()
print(item)
Output:
(1, 'Greg')
(2, 'Meera')
(3, 'Maureen')
(3, 'Mukhtar')Conclusion
That’s been our tutorial on the python priority queue.
We discussed how to implement a priority queue using a list data structure. Moreover, we showed how to make use of python’s own implementation of a priority queue.
We hope the tutorial was helpful to you in your own coding projects. Feel free to consider checking out our other tutorials
