5 Best Ways to Implement Priority Queue in Python

πŸ’‘ Problem Formulation: Priority queues are abstract data types that manage a set of records with totally-ordered keys (i.e., priorities) to allow for quick insertion of items and removal of the item with the highest priority. In Python, a typical use case could be scheduling tasks where the task with the highest urgency (priority) gets executed first. Let’s check input as a list of tuples where the first element is the task and the second is its priority, with the desired output allowing us to process tasks in priority order.

Method 1: Using the heapq Module

This method introduces the built-in Python heapq module, which implements a min-heap by default, meaning that the smallest element can be popped off first. To use it as a priority queue, invert the priority by multiplying it by -1 if you want the highest priority to be removed first. It’s efficient and easy to use for both smallest-first and largest-first priority queues.

Here’s an example:

import heapq

# A class to wrap tasks with reverse priority
class PriorityQueue:
    def __init__(self):
        self.heap = []
        self.count = 0

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

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

# Example usage
pq = PriorityQueue()
pq.push('task1', priority=3)
pq.push('task2', priority=1)
pq.push('task3', priority=2)
print(pq.pop())  # Should print task with highest priority

Output:

task1

In this example, PriorityQueue class is created with methods push and pop to add and remove tasks, respectively. It uses heapq.heappush and heapq.heappop to maintain the heap invariant. The count variable is used to ensure that tasks with the same priority will be popped in the order they were pushed.

Method 2: Using the queue.PriorityQueue Class

The queue.PriorityQueue class from Python’s standard library provides a way to implement a priority queue with built-in synchronization that makes it safe for concurrent access. It can be used in multi-threaded applications where multiple threads may be inserting or removing items from the priority queue at the same time.

Here’s an example:

from queue import PriorityQueue

pq = PriorityQueue()
pq.put((3, 'task1'))
pq.put((1, 'task2'))
pq.put((2, 'task3'))

while not pq.empty():
    print(pq.get()[1])  # Will print tasks in priority order

Output:

task2
task3
task1

This example shows a priority queue where tasks are inserted with a priority number. The PriorityQueue automatically sorts items so that when calling pq.get(), it retrieves the item with the lowest priority number, hence the reverse order in output.

Method 3: Using Sorted List

One can manually manage a list by maintaining its order after each insertion, effectively turning it into a priority queue. This method is straightforward but not recommended for larger datasets due to its inefficiency, as inserting an item into a list is an O(n) operation.

Here’s an example:

tasks = []

def add_task(task, priority):
    tasks.append((priority, task))
    tasks.sort(reverse=True)

def get_task():
    return tasks.pop()[1]

# Example usage
add_task('task1', 3)
add_task('task2', 1)
add_task('task3', 2)
print(get_task())  # Should print task with highest priority

Output:

task1

The function add_task appends a new task tuple to the list and then sorts it based on priority, with the highest priority first. get_task pops off the task with the highest priority from the end of the list.

Method 4: Using Object-Oriented Approach with Overridden Comparators

For a more structured approach, one can define a class for tasks where comparators are overridden to decide the sorting order based on priority. When using a list or any sortable container, Python will use these comparators to maintain a priority queue.

Here’s an example:

class Task:
    def __init__(self, task, priority):
        self.task = task
        self.priority = priority

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

tasks = []
tasks.append(Task('task1', 3))
tasks.append(Task('task2', 1))
tasks.sort(reverse=True)

print(tasks.pop().task)  # Should print task with highest priority

Output:

task1

Each Task object holds a task and its priority. The __lt__ method is overridden to compare tasks based on priority. When the list of Task objects is sorted, it uses this comparison to order the tasks correctly. The highest priority task is retrieved using pop method.

Bonus One-Liner Method 5: Using a List Comprehension and the max Function

For simplicity in smaller or less frequent operations, one can use a combination of list comprehension and the max function to retrieve the highest priority task. This method does not maintain the priority queue but can be used to find a task with the highest priority efficiently in an unsorted list.

Here’s an example:

tasks = [('task1', 3), ('task2', 1), ('task3', 2)]
print(max(tasks, key=lambda x: x[1])[0])  # Finds the task with highest priority

Output:

task1

This one-liner finds the maximum element in the list of tuples tasks, using the key parameter to compare elements by the second item in each tuple (i.e., the priority). It retrieves the task with the highest priority.

Summary/Discussion

  • Method 1: heapq Module. Efficient for large datasets. Might require inverting priorities for max-heap functionality.
  • Method 2: queue.PriorityQueue. Thread-safe for concurrent operations. Slower because of locking mechanisms.
  • Method 3: Sorted List. Simple implementation. Inefficient for large datasets due to O(n) insertion complexity.
  • Method 4: Object-Oriented Approach. Provides structured and clear code. Requires additional code overhead for class definition.
  • Method 5: List Comprehension and max Function. Great for one-off calculations in small list. Does not maintain a queue.