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:

  1. If we dequeued items based on minimum priority, we have a min-priority queue.
  2. 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 Carlos

However, 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 queue

You 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 item
  • get(): Dequeues the front item
  • empty(): Checks if the priority queue is empty
  • full(): 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

Like this article? Follow us on Facebook and LinkedIn.