How to Implement a Priority Queue in Python

A priority queue is a special type of queue where each element has an associated priority. Unlike a regular queue that follows First-In-First-Out (FIFO) ordering, a priority queue serves elements based on their priority: items with higher (or lower) priority are dequeued first, regardless of when they were added.

Think of it like an emergency room: patients with more severe conditions are treated before those who arrived earlier with minor issues.

In this guide, you will learn what priority queues are, how they differ from regular queues, and how to implement them in Python using multiple approaches. These range from a simple list-based solution to Python's built-in heapq module and queue.PriorityQueue class.

Key Properties of a Priority Queue

• High-priority elements are dequeued before low-priority ones.
• If two elements share the same priority, they are dequeued in the order they were inserted (FIFO behavior).
• Dynamic ordering: The queue automatically adjusts as new elements are added or removed.

Priority Queue vs. Regular Queue

Understanding the difference between these two structures helps you choose the right one for tasks like scheduling, resource management, or graph algorithms.

Types of Priority Queues

There are two main types of priority queues:

• Max Priority Queue: The element with the highest priority (largest value) is dequeued first. Useful when you need to process the most important or largest element first.
• Min Priority Queue: The element with the lowest priority (smallest value) is dequeued first. Ideal for problems like finding the shortest path or processing the least urgent task first.

Implementing a Priority Queue Using a List

The simplest way to implement a priority queue is with a Python list. This approach is easy to understand but not efficient for large datasets because finding the highest-priority element requires scanning the entire list on every deletion (resulting in O(n) time complexity).

def insert(queue, data):
    queue.append(data)

def delete(queue):
    if is_empty(queue):
        raise IndexError("Priority queue is empty.")
    max_index = 0
    for i in range(1, len(queue)):
        if queue[i] > queue[max_index]:
            max_index = i
    return queue.pop(max_index)

def is_empty(queue):
    return len(queue) == 0

if __name__ == "__main__":
    pq = []

    insert(pq, 12)
    insert(pq, 1)
    insert(pq, 14)
    insert(pq, 7)

    print("Queue:", pq)
    print("Removed elements (highest priority first):")
    while not is_empty(pq):
        print(delete(pq))

Output:

Queue: [12, 1, 14, 7]
Removed elements (highest priority first):
14
12
7
1

How it works:

• insert() appends the element to the end of the list.
• delete() scans the list to find the element with the highest value (max priority), removes it, and returns it.
• is_empty() checks if the list is empty.
• The while loop repeatedly removes and prints the highest-priority element until the queue is empty.

Implementing a Priority Queue Using heapq

Python's built-in heapq module provides an efficient min-heap implementation. Insertion and deletion both run in O(log n) time, making this the preferred approach for most use cases.

By default, heapq implements a min priority queue (smallest value has highest priority).

import heapq

pq = []

heapq.heappush(pq, 12)
heapq.heappush(pq, 1)
heapq.heappush(pq, 14)
heapq.heappush(pq, 7)

print("Removed elements (lowest priority first):")
while pq:
    print(heapq.heappop(pq))

Output:

Removed elements (lowest priority first):
1
7
12
14

Creating a Max Priority Queue with heapq

Since heapq only provides a min-heap, you can simulate a max priority queue by negating the values:

import heapq

pq = []

for value in [12, 1, 14, 7]:
    heapq.heappush(pq, -value)  # Negate to simulate max-heap

print("Removed elements (highest priority first):")
while pq:
    print(-heapq.heappop(pq))   # Negate again to get original value

Output:

Removed elements (highest priority first):
14
12
7
1

Using heapq with Custom Priority

In real-world scenarios, you often need to associate a priority with a task or data object. You can do this by pushing tuples where the first element is the priority:

import heapq

pq = []

heapq.heappush(pq, (2, "Medium priority task"))
heapq.heappush(pq, (1, "High priority task"))
heapq.heappush(pq, (3, "Low priority task"))

print("Processing tasks:")
while pq:
    priority, task = heapq.heappop(pq)
    print(f"  Priority {priority}: {task}")

Output:

Processing tasks:
  Priority 1: High priority task
  Priority 2: Medium priority task
  Priority 3: Low priority task

Implementing a Priority Queue Using queue.PriorityQueue

Python's queue module provides a thread-safe PriorityQueue class. This is the best choice when multiple threads need to access the queue concurrently.

from queue import PriorityQueue

pq = PriorityQueue()

pq.put((2, "Medium priority task"))
pq.put((1, "High priority task"))
pq.put((3, "Low priority task"))

print("Processing tasks:")
while not pq.empty():
    priority, task = pq.get()
    print(f"  Priority {priority}: {task}")

Output:

Processing tasks:
  Priority 1: High priority task
  Priority 2: Medium priority task
  Priority 3: Low priority task

Comparison of Approaches

Common Mistake: Comparing Non-Comparable Elements

A frequent error when using heapq or PriorityQueue with tuples occurs when two items share the same priority and the secondary element cannot be compared:

import heapq

pq = []
heapq.heappush(pq, (1, {"name": "Task A"}))
heapq.heappush(pq, (1, {"name": "Task B"}))  # Same priority!

Output:

TypeError: '<' not supported between instances of 'dict' and 'dict'

The fix: Add a unique tie-breaking counter as the second element:

import heapq

pq = []
heapq.heappush(pq, (1, 0, {"name": "Task A"}))
heapq.heappush(pq, (1, 1, {"name": "Task B"}))

while pq:
    priority, _, task = heapq.heappop(pq)
    print(f"  Priority {priority}: {task}")

Output:

  Priority 1: {'name': 'Task A'}
  Priority 1: {'name': 'Task B'}

Applications of Priority Queues

Priority queues are fundamental in many areas of computer science and software engineering:

• Task Scheduling (Operating Systems): This manages processes by priority, executing high-priority tasks first in real-time systems.
• Dijkstra's Shortest Path Algorithm: This uses a min priority queue to repeatedly select the nearest unvisited node.
• Huffman Encoding (Data Compression): This combines the least frequent symbols using a priority queue to build an optimal encoding tree.
• A* Search Algorithm (Pathfinding): This prioritizes nodes based on estimated total cost to find the shortest path in navigation or games.
• Merging K Sorted Lists: This efficiently merges multiple sorted lists by always selecting the smallest available element.
• Event-Driven Simulation: This processes events in chronological order based on scheduled timestamps.

Summary

Python offers several ways to implement priority queues, each suited to different scenarios:

• Use a simple list for learning purposes or very small datasets.
• Use heapq for an efficient, lightweight implementation in single-threaded applications.
• Use queue.PriorityQueue when you need thread-safe access from multiple threads.

For most applications, heapq strikes the best balance between simplicity and performance. Remember to use tie-breaking counters when working with elements that may share the same priority to avoid comparison errors.