Python Priority Queue: Comprehensive Guide

A priority queue is a special type of queue in which each element has a priority associated with it. Elements with higher priority are dequeued before elements with lower priority, regardless of their order in the queue. Priority queues are widely used in algorithms like Dijkstra’s shortest path and scheduling tasks.

In Python, priority queues can be implemented using various methods, including:

heapq module (the most common way)
queue.PriorityQueue class from the queue module
Custom implementations using data structures like binary heaps or sorted lists

By the end of this guide, you will have a complete understanding of how to implement and use a Python priority queue effectively.

What is a Priority Queue?

A priority queue is a data structure where each element has a priority associated with it. In a priority queue:

Elements with higher priority are dequeued first.
If two elements have the same priority, they are dequeued in the order they were added (FIFO for elements with equal priority).

In Python, you can think of a priority queue as a queue that dequeues the smallest or largest element first, depending on the implementation. This is useful in scenarios where you need to process elements based on their importance or urgency.

Real-World Examples of Priority Queues

Task scheduling: Tasks with higher priority are executed first.
Pathfinding algorithms: In Dijkstra’s algorithm, nodes with the smallest distance are processed first.
Event-driven simulations: Events with the nearest timestamp are processed first.

Implementing a Priority Queue in Python

1. Using `heapq` Module

Python’s heapq module provides a simple way to implement a priority queue using a binary heap. Heaps are binary trees where the parent node is smaller than its children (min-heap) or larger (max-heap). By default, Python implements a min-heap using heapq.

Basic Operations with `heapq`:

heapq.heappush(heap, item): Adds an item to the heap.
heapq.heappop(heap): Removes and returns the smallest item from the heap.
heapq.heapify(x): Transforms a list into a heap.

Example: Simple Priority Queue Using `heapq`

import heapq

# Create an empty heap
priority_queue = []

# Add elements to the priority queue
heapq.heappush(priority_queue, (2, "task 2"))
heapq.heappush(priority_queue, (1, "task 1"))
heapq.heappush(priority_queue, (3, "task 3"))

# Remove elements based on priority (smallest first)
while priority_queue:
    priority, task = heapq.heappop(priority_queue)
    print(f"Processing {task} with priority {priority}")

Output:

Processing task 1 with priority 1
Processing task 2 with priority 2
Processing task 3 with priority 3

In this example, tasks are processed based on their priority, with the lowest priority number being dequeued first.

2. Using `queue.PriorityQueue` Class

Python’s queue.PriorityQueue is a thread-safe class for implementing priority queues. It is part of the queue module and is useful in multi-threaded environments where you need to coordinate access to shared resources.

Example: Priority Queue Using `queue.PriorityQueue`

from queue import PriorityQueue

# Create a PriorityQueue object
pq = PriorityQueue()

# Add elements to the priority queue
pq.put((2, "task 2"))
pq.put((1, "task 1"))
pq.put((3, "task 3"))

# Remove elements based on priority
while not pq.empty():
    priority, task = pq.get()
    print(f"Processing {task} with priority {priority}")

Output:

Processing task 1 with priority 1
Processing task 2 with priority 2
Processing task 3 with priority 3

In this case, the queue.PriorityQueue class provides a thread-safe implementation with the same priority-based ordering.

Customizing Priority in a Python Priority Queue

By default, Python’s heapq and queue.PriorityQueue process the smallest priority values first (min-heap). If you want to implement a max-heap (i.e., process the largest values first), you can negate the priority values when inserting items.

Example: Max-Heap with `heapq`

import heapq

# Create an empty heap
priority_queue = []

# Add elements with negative priority to simulate a max-heap
heapq.heappush(priority_queue, (-2, "task 2"))
heapq.heappush(priority_queue, (-1, "task 1"))
heapq.heappush(priority_queue, (-3, "task 3"))

# Remove elements (largest priority first)
while priority_queue:
    priority, task = heapq.heappop(priority_queue)
    print(f"Processing {task} with priority {-priority}")

Output:

Processing task 3 with priority 3
Processing task 2 with priority 2
Processing task 1 with priority 1

In this example, by negating the priority values, we simulate a max-heap where the highest priority tasks are processed first.

Common Use Cases for Python Priority Queue

1. Dijkstra’s Algorithm for Shortest Path

In graph algorithms like Dijkstra’s, you use a priority queue to always process the node with the smallest distance from the starting node.

Example:

import heapq

def dijkstra(graph, start):
    priority_queue = []
    heapq.heappush(priority_queue, (0, start))
    distances = {node: float('infinity') for node in graph}
    distances[start] = 0

    while priority_queue:
        current_distance, current_node = heapq.heappop(priority_queue)

        if current_distance > distances[current_node]:
            continue

        for neighbor, weight in graph[current_node]:
            distance = current_distance + weight

            if distance &lt; distances[neighbor]:
                distances[neighbor] = distance
                heapq.heappush(priority_queue, (distance, neighbor))

    return distances

2. Task Scheduling

A priority queue is often used in operating systems to schedule tasks based on their priority. Tasks with higher priority are executed first.

Example:

import heapq

tasks = [(2, "Email"), (1, "Coding"), (3, "Meeting")]
priority_queue = []

for task in tasks:
    heapq.heappush(priority_queue, task)

while priority_queue:
    priority, task = heapq.heappop(priority_queue)
    print(f"Performing task: {task}")

Best Practices for Working with Priority Queues

1. Use `heapq` for Simplicity and Efficiency

For most use cases where you need a priority queue, heapq is the most efficient and simple way to implement it. It provides all the basic operations for priority queues with good performance characteristics (insertion and extraction in O(log n)).

2. Use `queue.PriorityQueue` for Multi-Threaded Applications

If you’re working in a multi-threaded environment, use queue.PriorityQueue from the queue module to ensure thread safety.

3. Handle Custom Priorities with Care

When working with custom priorities (e.g., using max-heaps or complex sorting rules), ensure that you use a consistent priority scheme (e.g., negating priorities for a max-heap). Be careful with how you define your priority values to avoid bugs.

Common Pitfalls When Using Priority Queues

1. Confusing Min-Heap and Max-Heap

By default, Python’s heapq implements a min-heap, meaning that the smallest element is processed first. If you want to process the largest elements first (max-heap), you need to negate the priority values or implement custom logic to reverse the order.

2. Forgetting Tuple Structure in `heapq`

When using heapq, you must push tuples into the heap where the first element of the tuple is the priority, and the second element is the data. Failing to do this can lead to errors or incorrect behavior.

Incorrect:

heapq.heappush(priority_queue, "task")

Correct:

heapq.heappush(priority_queue, (1, "task"))

3. Ignoring Thread Safety

If you are working in a multi-threaded environment, always use queue.PriorityQueue instead of heapq, as heapq is not thread-safe. queue.PriorityQueue ensures that elements are safely accessed by multiple threads.

Summary of Key Concepts

A priority queue is a data structure where each element has a priority, and the element with the highest (or lowest) priority is dequeued first.
Python provides two main ways to implement a priority queue:
heapq module: Simple, efficient, and non-thread-safe implementation using heaps.
queue.PriorityQueue: Thread-safe class for use in multi-threaded applications.
By default, heapq implements a min-heap (smallest elements dequeued first), but you can implement a max-heap by negating priority values.
Priority queues are commonly used in algorithms like Dijkstra’s shortest path and task scheduling.

Exercises

Basic Python Priority Queue: Write a Python program that uses heapq to implement a priority queue where the highest priority items are processed first.
Task Scheduling: Create a task scheduler using queue.PriorityQueue that allows users to add tasks with priorities and processes tasks based on their priority in a multi-threaded environment.
Dijkstra’s Algorithm: Implement Dijkstra’s shortest path algorithm using a priority queue to find the shortest path in a weighted graph.

Lightning bolt and Python code snippet with "LEARN PYTHON PROGRAMMING MASTERCLASS" in blocky caps

Check out our FREE Learn Python Programming Masterclass to hone your skills or learn from scratch.

The course covers everything from first principles to Graphical User Interfaces and Machine Learning

Refer to the official Python priority queue documentation here.

FAQ

Q1: Can I store complex objects in a priority queue instead of just numbers or strings?

A1: Yes, you can store any type of object in a priority queue, but you need to define a priority for each object. Typically, you store tuples where the first element is the priority and the second is the object itself. If the object has a more complex structure, you can still define custom comparisons to sort them based on priority.

Example:

import heapq

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

    def __repr__(self):
        return f"{self.name} (priority: {self.priority})"

priority_queue = []
heapq.heappush(priority_queue, (2, Task("Email", 2)))
heapq.heappush(priority_queue, (1, Task("Coding", 1)))

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

Q2: How does `heapq` ensure that the smallest element is always dequeued first?

A2: The heapq module in Python uses a binary heap data structure, specifically a min-heap by default. In a binary heap, the parent node is always smaller than its child nodes. This property ensures that the smallest element is always at the root, allowing heapq.heappop() to remove the smallest element in O(log n) time. The heap is then restructured to maintain this property.

Q3: Can I use a priority queue for max-priority elements instead of min-priority?

A3: Yes, you can simulate a max-heap (where the largest elements are dequeued first) by negating the priority values when you push elements into the heap. By storing negative values, the smallest number in terms of magnitude (i.e., most negative) becomes the largest when processed.

Example:

import heapq

priority_queue = []
heapq.heappush(priority_queue, (-2, "task 2"))
heapq.heappush(priority_queue, (-1, "task 1"))
heapq.heappush(priority_queue, (-3, "task 3"))

while priority_queue:
    priority, task = heapq.heappop(priority_queue)
    print(f"Processing {task} with priority {-priority}")

Q4: What happens if two tasks have the same priority?

A4: When two tasks have the same priority, heapq maintains their original order as they were inserted into the priority queue. This behavior is consistent with the concept of stability in sorting, where elements with equal priority are dequeued in the same order in which they were added (FIFO for equal priority).

Example:

import heapq

priority_queue = []
heapq.heappush(priority_queue, (1, "task 1"))
heapq.heappush(priority_queue, (1, "task 2"))

while priority_queue:
    priority, task = heapq.heappop(priority_queue)
    print(f"Processing {task} with priority {priority}")

Output:

Processing task 1 with priority 1
Processing task 2 with priority 1

Q5: Is `heapq` thread-safe?

A5: No, the heapq module is not thread-safe. If you need to use a priority queue in a multi-threaded environment, you should use the queue.PriorityQueue class from the queue module, which provides thread-safe operations on priority queues.

Example using `queue.PriorityQueue`:

from queue import PriorityQueue

pq = PriorityQueue()
pq.put((1, "task 1"))
pq.put((2, "task 2"))

Q6: Can I change the priority of an item already in the priority queue?

A6: There is no built-in way to change the priority of an item already in the priority queue. A common workaround is to remove and reinsert the item with the updated priority, or use a special approach by marking the old item as invalid and adding a new entry with the updated priority.

Example: Marking an Item as Invalid

import heapq

priority_queue = []
entry_finder = {}  # Mapping of tasks to entries
REMOVED = '&lt;removed-task>'  # Placeholder for removed tasks
counter = 0  # Unique sequence count

def add_task(task, priority):
    entry = [priority, counter, task]
    entry_finder[task] = entry
    heapq.heappush(priority_queue, entry)

def remove_task(task):
    entry = entry_finder.pop(task)
    entry[-1] = REMOVED

def pop_task():
    while priority_queue:
        priority, _, task = heapq.heappop(priority_queue)
        if task is not REMOVED:
            del entry_finder[task]
            return task
    raise KeyError('pop from an empty priority queue')

# Example usage:
add_task('task1', 1)
add_task('task2', 2)
remove_task('task1')  # Mark task1 as removed
add_task('task1', 3)  # Add task1 again with a new priority

print(pop_task())  # Output: task2
print(pop_task())  # Output: task1

Q7: What is the time complexity of operations in `heapq` and `queue.PriorityQueue`?

A7: Both heapq and queue.PriorityQueue are based on binary heaps, so their time complexity for the main operations is:

heappush() or put(): O(log n)
heappop() or get(): O(log n)
heapify() (transforming a list into a heap): O(n)

These operations are efficient and scale well with the size of the heap.

Q8: Can I create a priority queue that uses different types of objects (e.g., strings and numbers) as priority?

A8: No, you cannot mix different types like strings and numbers as priorities in a Python priority queue. Python will raise a TypeError because it cannot compare different types. If you need to handle mixed types, you can define a custom comparison or convert the priorities into a uniform type (e.g., by using integers).

Example of Type Error:

import heapq

priority_queue = []
# This will raise a TypeError because Python can't compare int and str
heapq.heappush(priority_queue, (1, "task 1"))
heapq.heappush(priority_queue, ("high", "task 2"))

To avoid this issue, use consistent data types for priority values.

Q9: Can a priority queue contain duplicate elements?

A9: Yes, priority queues can contain duplicate elements, both in terms of the priorities and the tasks. However, when you pop elements, they will be returned in the order they were added if their priority is the same. You can also handle duplicate entries manually if needed.

Example:

import heapq

priority_queue = []
heapq.heappush(priority_queue, (1, "task"))
heapq.heappush(priority_queue, (1, "task"))

while priority_queue:
    print(heapq.heappop(priority_queue))

In this case, the same task is added twice with the same priority, and both will be processed.

Q10: Can I use a priority queue to sort elements?

A10: Yes, you can use a priority queue (min-heap) to sort elements efficiently. By inserting elements into the heap and then removing them in order, you can implement heap sort, which sorts elements in O(n log n) time.

Example: Sorting with `heapq`

import heapq

numbers = [5, 1, 8, 3, 2]
heapq.heapify(numbers)  # Transform the list into a heap

sorted_numbers = [heapq.heappop(numbers) for _ in range(len(numbers))]
print(sorted_numbers)  # Output: [1, 2, 3, 5, 8]

In this example, the elements are sorted using a priority queue.

Table of Contents

What is a Priority Queue?

Real-World Examples of Priority Queues

Implementing a Priority Queue in Python

1. Using heapq Module

Basic Operations with heapq:

Example: Simple Priority Queue Using heapq

Output:

2. Using queue.PriorityQueue Class

Example: Priority Queue Using queue.PriorityQueue

Output:

Customizing Priority in a Python Priority Queue

Example: Max-Heap with heapq

Output:

Common Use Cases for Python Priority Queue

1. Dijkstra’s Algorithm for Shortest Path

Example:

2. Task Scheduling

Example:

Best Practices for Working with Priority Queues

1. Use heapq for Simplicity and Efficiency

2. Use queue.PriorityQueue for Multi-Threaded Applications

3. Handle Custom Priorities with Care

Common Pitfalls When Using Priority Queues

1. Confusing Min-Heap and Max-Heap

2. Forgetting Tuple Structure in heapq

Incorrect:

Correct:

3. Ignoring Thread Safety

Summary of Key Concepts

Exercises

FAQ

Q1: Can I store complex objects in a priority queue instead of just numbers or strings?

Example:

Q2: How does heapq ensure that the smallest element is always dequeued first?

Q3: Can I use a priority queue for max-priority elements instead of min-priority?

Example:

Q4: What happens if two tasks have the same priority?

Example:

Output:

Q5: Is heapq thread-safe?

Example using queue.PriorityQueue:

Example: Marking an Item as Invalid

Q7: What is the time complexity of operations in heapq and queue.PriorityQueue?

Q8: Can I create a priority queue that uses different types of objects (e.g., strings and numbers) as priority?

Example of Type Error:

Q9: Can a priority queue contain duplicate elements?

Example:

Q10: Can I use a priority queue to sort elements?

Example: Sorting with heapq

Similar Posts

1. Using `heapq` Module

Basic Operations with `heapq`:

Example: Simple Priority Queue Using `heapq`

2. Using `queue.PriorityQueue` Class

Example: Priority Queue Using `queue.PriorityQueue`

Example: Max-Heap with `heapq`

1. Use `heapq` for Simplicity and Efficiency

2. Use `queue.PriorityQueue` for Multi-Threaded Applications

2. Forgetting Tuple Structure in `heapq`

Q2: How does `heapq` ensure that the smallest element is always dequeued first?

Q5: Is `heapq` thread-safe?

Example using `queue.PriorityQueue`:

Q7: What is the time complexity of operations in `heapq` and `queue.PriorityQueue`?

Example: Sorting with `heapq`