How to implement a priority queue in Python

A priority queue is a data structure that stores elements with associated priorities. Python's heapq module offers an efficient way to build one for tasks that depend on element order.

In this article, you will explore implementation techniques and real world applications. We'll cover practical tips and debugging advice to help you select the right approach for your project.

Using heapq module for a basic priority queue

import heapq

pq = []
heapq.heappush(pq, (2, "Medium priority task"))
heapq.heappush(pq, (1, "High priority task"))
heapq.heappush(pq, (3, "Low priority task"))
while pq:
   priority, task = heapq.heappop(pq)
   print(f"Processing {task} with priority {priority}")--OUTPUT--Processing High priority task with priority 1
Processing Medium priority task with priority 2
Processing Low priority task with priority 3

The heapq module transforms a standard Python list into a min-heap, which keeps the item with the lowest value at the front. By using tuples like (priority, task), you're telling heapq to use the first element for sorting. This is why a lower number means a higher priority in the queue.

• The heappush function adds an item while maintaining the heap structure.
• The heappop function removes and returns the smallest item, ensuring the task with priority 1 is always processed first.

Alternative implementations

While the heapq module is a solid default, Python's flexibility offers other ways to create priority queues for more specialized use cases.

Using queue.PriorityQueue class for thread-safe operations

from queue import PriorityQueue

pq = PriorityQueue()
pq.put((2, "Medium priority task"))
pq.put((1, "High priority task"))
pq.put((3, "Low priority task"))
while not pq.empty():
   priority, task = pq.get()
   print(f"Processing {task} with priority {priority}")--OUTPUT--Processing High priority task with priority 1
Processing Medium priority task with priority 2
Processing Low priority task with priority 3

The queue.PriorityQueue class is built for concurrent applications. It’s thread-safe, meaning it automatically handles the locking required when multiple threads access the queue at once. This prevents data corruption without you needing to manage locks manually.

• You use methods like put() to add items and get() to retrieve them in priority order.
• This object-oriented approach provides a cleaner, more encapsulated interface than the function-based heapq module, which is ideal for multi-threaded programs.

Creating a priority queue with a sorted list

tasks = []
tasks.append((2, "Medium priority task"))
tasks.append((1, "High priority task"))
tasks.append((3, "Low priority task"))
tasks.sort()
for priority, task in tasks:
   print(f"Processing {task} with priority {priority}")--OUTPUT--Processing High priority task with priority 1
Processing Medium priority task with priority 2
Processing Low priority task with priority 3

A simple list can also function as a priority queue, especially when you can add all items at once. After appending your tasks, a single call to the sort() method arranges them in priority order. This approach is direct and leverages Python’s built-in sorting.

• The main drawback is performance. If you add new items frequently, you must re-sort the entire list each time, which is inefficient.
• This method is best suited for batch processing where all tasks are known beforehand.

Using bisect module for insertion in order

import bisect

tasks = []
bisect.insort(tasks, (2, "Medium priority task"))
bisect.insort(tasks, (1, "High priority task"))
bisect.insort(tasks, (3, "Low priority task"))
for priority, task in tasks:
   print(f"Processing {task} with priority {priority}")--OUTPUT--Processing High priority task with priority 1
Processing Medium priority task with priority 2
Processing Low priority task with priority 3

The bisect module helps you maintain a sorted list without the overhead of re-sorting it completely. The bisect.insort() function efficiently finds the correct position for a new item and inserts it, keeping the list ordered with every addition.

• This approach is more performant than appending and then sorting if you're adding items incrementally.
• While insertion is efficient, removing the highest-priority item from the front of the list is slow, making heapq a better choice for classic priority queue behavior.

Advanced techniques

While tuples work for simple cases, you can unlock more power and clarity by creating custom classes to manage priority and data together.

Implementing a custom priority queue class

import heapq

class PriorityQueue:
   def __init__(self):
       self.heap = []
       self.entry_count = 0
   
   def push(self, priority, task):
       heapq.heappush(self.heap, (priority, self.entry_count, task))
       self.entry_count += 1
   
   def pop(self):
       priority, _, task = heapq.heappop(self.heap)
       return priority, task

pq = PriorityQueue()
pq.push(2, "Medium priority task")
pq.push(1, "High priority task")
pq.push(3, "Low priority task")
while pq.heap:
   priority, task = pq.pop()
   print(f"Processing {task} with priority {priority}")--OUTPUT--Processing High priority task with priority 1
Processing Medium priority task with priority 2
Processing Low priority task with priority 3

Wrapping heapq logic in a custom class gives you a cleaner, more object-oriented interface. This approach encapsulates the heap operations within methods like push() and pop(), making your code more organized and reusable.

• The entry_count variable is the key addition here. It acts as a tie-breaker when priorities are identical.
• If two tasks have the same priority, the one added earlier will have a lower entry_count and will be processed first. This creates a stable, first-in, first-out order for items of equal importance.

Using heapq with custom objects via __lt__ method

import heapq

class Task:
   def __init__(self, priority, description):
       self.priority = priority
       self.description = description
   
   def __lt__(self, other):
       return self.priority < other.priority

tasks = []
heapq.heappush(tasks, Task(2, "Medium priority task"))
heapq.heappush(tasks, Task(1, "High priority task"))
heapq.heappush(tasks, Task(3, "Low priority task"))
while tasks:
   task = heapq.heappop(tasks)
   print(f"Processing {task.description} with priority {task.priority}")--OUTPUT--Processing High priority task with priority 1
Processing Medium priority task with priority 2
Processing Low priority task with priority 3

Using a custom class like Task makes your code more expressive than using tuples. It bundles priority and data together in a clear, object-oriented way. The magic happens when you define how Python should compare your custom objects, which allows heapq to sort them correctly.

• By implementing the __lt__ method, you're telling Python what the less-than operator (<) means for a Task.
• The heapq module automatically uses this method to compare items and maintain the heap structure, sorting tasks based on their priority attribute without any extra code.

Using dataclasses for efficient priority queue items

import heapq
from dataclasses import dataclass, field

@dataclass(order=True)
class PrioritizedTask:
   priority: int
   task: str = field(compare=False)

pq = []
heapq.heappush(pq, PrioritizedTask(2, "Medium priority task"))
heapq.heappush(pq, PrioritizedTask(1, "High priority task"))
heapq.heappush(pq, PrioritizedTask(3, "Low priority task"))
while pq:
   item = heapq.heappop(pq)
   print(f"Processing {item.task} with priority {item.priority}")--OUTPUT--Processing High priority task with priority 1
Processing Medium priority task with priority 2
Processing Low priority task with priority 3

Python's dataclasses offer a clean and modern way to create classes for priority queue items. The @dataclass(order=True) decorator automatically generates the necessary comparison methods, so you don't have to write them yourself.

• The fields are compared in the order they are defined, so priority is used for sorting.
• By setting task: str = field(compare=False), you tell the dataclass to ignore the task description during comparisons. This ensures sorting is based only on priority, which is exactly what you want for a priority queue.

Move faster with Replit

Replit is an AI-powered development platform that transforms natural language into working applications. Describe what you want to build, and Replit Agent creates it—complete with databases, APIs, and deployment.

For the priority queue implementations we've explored, Replit Agent can turn them into production-ready tools:

• Build a customer support ticket system that automatically surfaces the most critical issues first.
• Create a background job processor that handles urgent tasks like password resets before routine operations.
• Deploy a real-time notification dispatcher that prioritizes security alerts over promotional messages.

Try it for yourself. Describe your application idea, and Replit Agent will write the code, test it, and deploy it automatically.

Common errors and challenges

Implementing priority queues in Python is straightforward, but a few common issues can trip you up if you're not prepared for them.

• Handling items with identical priorities. When two items in a heapq queue share a priority, Python tries to compare the next element in their tuples to break the tie. If that second element is a complex object that can't be compared, your code will raise a TypeError. The best fix is to add a unique entry counter as a tie-breaker, ensuring a stable, first-in-first-out order.
• Forgetting comparison methods for custom objects. If you use custom objects in a heap, you must tell Python how to compare them. A frequent mistake is forgetting to implement the __lt__ (less-than) method, which leaves heapq with no way to sort your objects and results in a TypeError.
• Creating a max-heap. Python’s heapq module is a min-heap, so it always returns the smallest item. To get the largest item instead, you can simulate a max-heap by pushing the negative value of each priority onto the heap. When you pop an item, just multiply its priority by -1 to restore the original value.

Handling items with identical priorities in heapq

When items in a heapq share a priority, Python breaks the tie by comparing the next element in the tuple. This can cause unpredictable sorting based on data you didn't intend to use for ordering, as the following code demonstrates.

import heapq

pq = []
heapq.heappush(pq, (1, "Task A"))
heapq.heappush(pq, (1, "Task B"))
heapq.heappush(pq, (1, "Task C"))

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

Since all priorities are 1, heapq defaults to sorting the tasks alphabetically by their string descriptions. This isn't the first-in, first-out order you might expect. The next example shows how to fix this for stable sorting.

import heapq

pq = []
counter = 0

heapq.heappush(pq, (1, counter, "Task A"))
counter += 1
heapq.heappush(pq, (1, counter, "Task B"))
counter += 1
heapq.heappush(pq, (1, counter, "Task C"))
counter += 1

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

The fix is to add a unique, auto-incrementing counter to each tuple. This ensures stable sorting when priorities are identical.

• The heapq module uses the counter as a tie-breaker, falling back on it when priorities match.
• Since the counter value increases with each new item, tasks added earlier are guaranteed to be processed first. This creates a reliable first-in, first-out (FIFO) order for items of equal importance.

Forgetting to implement __lt__ for custom objects in heaps

When using custom objects in a priority queue, Python's heapq module must know how to compare them. If you don't define a comparison method like __lt__, the heap can't sort your objects, which triggers a TypeError. The code below demonstrates this exact issue.

import heapq

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

tasks = []
heapq.heappush(tasks, Task(2, "Medium priority task"))
heapq.heappush(tasks, Task(1, "High priority task"))
heapq.heappush(tasks, Task(3, "Low priority task"))

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

The error occurs when heapq tries to sort the Task objects but finds no instructions for how to compare them. This ambiguity results in a TypeError. The corrected code below adds the necessary comparison logic.

import heapq

class Task:
   def __init__(self, priority, description):
       self.priority = priority
       self.description = description
   
   def __lt__(self, other):
       return self.priority < other.priority

tasks = []
heapq.heappush(tasks, Task(2, "Medium priority task"))
heapq.heappush(tasks, Task(1, "High priority task"))
heapq.heappush(tasks, Task(3, "Low priority task"))

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

The fix is to implement the __lt__ (less-than) method in your custom class. This method tells Python how to compare two objects. By adding logic like return self.priority < other.priority, you give heapq clear instructions to sort tasks based on their priority attribute.

With this comparison defined, heapq can correctly maintain the heap order, preventing the TypeError. This ensures the task with the lowest priority is always processed first.

Creating a max heap with heapq

Python's heapq module is a min-heap, so it always returns the smallest item. This is perfect for ascending priority but creates a problem when you need a max-heap to process the largest items first. The code below demonstrates this default behavior.

import heapq

pq = []
heapq.heappush(pq, 3)
heapq.heappush(pq, 1)
heapq.heappush(pq, 4)
heapq.heappush(pq, 2)

print("Items in descending order:")
while pq:
   item = heapq.heappop(pq)
   print(item)

The code intends to print items in descending order, but heapq.heappop() always extracts the smallest value. This produces an ascending sequence—the opposite of the desired outcome. The following example shows how to reverse this behavior.

import heapq

pq = []
heapq.heappush(pq, -3)
heapq.heappush(pq, -1)
heapq.heappush(pq, -4)
heapq.heappush(pq, -2)

print("Items in descending order:")
while pq:
   item = -heapq.heappop(pq)
   print(item)

You can simulate a max-heap by tricking the heapq module. The solution is to push the negative value of each priority onto the heap. Since heapq is a min-heap, it treats the largest original number—now the smallest negative—as the highest priority item.

• When you call heappop(), just negate the result to restore its original value.
• It's a simple workaround for when you need to process items in descending order.

Real-world applications

Beyond the implementation details and common errors, priority queues are the engine behind powerful applications like task schedulers and pathfinding algorithms.

Using heapq for task scheduling with timestamps

A priority queue is a natural fit for task scheduling, allowing you to use timestamps as priorities to process events in chronological order.

import heapq

# Tasks with priority based on scheduled execution time (in minutes from now)
schedule = []
heapq.heappush(schedule, (5, "Send email notification"))
heapq.heappush(schedule, (2, "Update user profile"))
heapq.heappush(schedule, (10, "Generate report"))

# Process tasks in order of scheduled time
while schedule:
   minutes, task = heapq.heappop(schedule)
   print(f"Executing: {task} (scheduled in {minutes} minutes)")

This example demonstrates how heapq efficiently organizes time-based events. By pushing tuples like (minutes, task) onto the heap, you're using the minutes value as the sorting key.

• The heappush function ensures the task with the lowest minutes value is always ready to be accessed.
• The while loop repeatedly calls heappop to retrieve and process the next available task, guaranteeing the "Update user profile" task runs first since it's scheduled in just two minutes.

Implementing Dijkstra's algorithm with heapq

The heapq module is also essential for pathfinding algorithms like Dijkstra's, where a priority queue is used to track and explore the shortest available routes in a graph.

import heapq

def shortest_path(graph, start, end):
   queue = [(0, start, [])]
   visited = set()
   
   while queue:
       (cost, node, path) = heapq.heappop(queue)
       
       if node == end:
           return path + [node], cost
       
       if node in visited:
           continue
           
       visited.add(node)
       
       for neighbor, weight in graph[node].items():
           if neighbor not in visited:
               heapq.heappush(queue, (cost + weight, neighbor, path + [node]))
   
   return [], float('infinity')

# Map with cities and distances
graph = {
   'A': {'B': 2, 'C': 5},
   'B': {'A': 2, 'C': 1, 'D': 3},
   'C': {'A': 5, 'B': 1, 'D': 2},
   'D': {'B': 3, 'C': 2}
}

path, distance = shortest_path(graph, 'A', 'D')
print(f"Path: {' -> '.join(path)}, Distance: {distance}")

This function, shortest_path, calculates the most efficient route through a network represented by the graph. It uses a priority queue to keep track of paths, always exploring the one with the lowest cumulative cost first. A visited set prevents the algorithm from getting stuck in loops by ensuring each node is processed only once.

• The while loop repeatedly pulls the shortest path found so far from the queue using heapq.heappop.
• It then adds the node's unvisited neighbors to the queue with their updated travel costs, continuing until the destination is reached.

Get started with Replit

Turn your knowledge into a real tool. Tell Replit Agent to "build a support ticket system that prioritizes urgent issues" or "create a task scheduler that runs jobs based on timestamps."

The agent writes the code, tests for errors, and deploys your application automatically. Start building with Replit.