How to implement a priority queue in Python

A priority queue is a data structure that organizes elements by their priority. In Python, its implementation helps you handle tasks efficiently, so the most critical items are processed first.

In this article, you'll explore several implementation techniques and practical tips for Python. We'll also cover real-world applications and provide solutions for common bugs, so you can confidently use priority queues in your projects.

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

Python's heapq module provides an efficient min-heap implementation, which is perfect for a priority queue. It operates directly on a list, using tuples like (priority, task) to organize data. The heap algorithm automatically uses the first element of the tuple for sorting, meaning a lower number signifies a higher priority.

The core logic relies on two functions:

• heapq.heappush() adds an item while maintaining the heap property.
• heapq.heappop() removes and returns the item with the smallest priority value, ensuring the most critical task is always processed first.

Alternative implementations

While the heapq module is a solid default, Python also provides the thread-safe queue.PriorityQueue class and tools like bisect for custom implementations.

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 your go-to for multithreaded applications where different threads need to access the queue safely. It handles all the locking mechanisms behind the scenes to prevent race conditions. Unlike heapq, this is a class-based implementation, so you interact with an object instance.

• Use the put() method to add items.
• Retrieve the highest-priority item with get().
• Check if the queue has items left with the empty() method.

This approach offers a clean, object-oriented interface for managing prioritized tasks in concurrent 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 sorted list offers a straightforward, if less dynamic, way to manage priorities. You can add all your tasks to a list and then call the sort() method. Because Python sorts tuples based on their first element, your tasks are automatically arranged by priority number.

• This approach is ideal when you have a batch of tasks to process all at once.
• It becomes inefficient if you need to add new items frequently, as the entire list must be re-sorted after each addition.

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 after every addition. The bisect.insort() function finds the correct spot for a new item and inserts it, keeping your task list ordered by priority automatically.

• This is more efficient than a simple sorted list if you're adding tasks one by one.
• While insertion is fast, removing the highest-priority item is slower than with a heap, as it requires shifting all other elements in the list.

Advanced techniques

Building on these fundamentals allows you to handle more complex objects by creating custom classes and defining their priority with methods like __lt__.

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

Encapsulating heapq in a custom class gives you a cleaner interface and solves a tricky problem: handling items with the same priority. This implementation adds an entry_count to maintain a stable, first-in-first-out order for tasks of equal importance.

• The push method adds a unique entry_count to each item, creating a tuple like (priority, entry_count, task).
• This counter acts as a tie-breaker. If two tasks have the same priority, the one added first will be processed first, preventing potential comparison errors with complex task objects.

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

You can make heapq work directly with custom objects by defining how Python should compare them. By implementing the special __lt__ method (short for "less than") in your Task class, you're telling the heap algorithm which attribute to use for sorting. This makes your code more intuitive, as you can push objects directly without wrapping them in tuples.

• The __lt__ method is automatically called by heapq to determine an item's priority.
• In this example, the line return self.priority < other.priority sorts tasks by their numerical priority.
• This approach keeps your data neatly encapsulated and your main logic cleaner.

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 concise way to build objects for your priority queue. When you use the @dataclass(order=True) decorator, it automatically generates the necessary comparison methods, making your custom objects ready for sorting with heapq.

• This setup compares objects based on the fields you define, starting with priority.
• The key is using field(compare=False) for the task attribute. This tells Python to ignore the task string during sorting, ensuring only the numerical priority is used and preventing comparison errors.

Move faster with Replit

Replit is an AI-powered development platform that comes with all Python dependencies pre-installed, so you can skip setup and start coding instantly.

Instead of piecing together techniques, you can describe the app you want to build and let Agent 4 take it from idea to working product. It handles writing the code, managing databases, and deploying your application directly from your description.

• A real-time support ticket system that automatically prioritizes critical issues for customer service agents.
• A background job processor for a web app that handles urgent tasks like password resets before less important ones like sending newsletters.
• A resource scheduler for a cloud service that allocates CPU time to high-priority computations first.

Simply describe your app, and Replit will write the code, test it, and fix issues automatically, all within your browser.

Common errors and challenges

Even with the right tools, you might run into a few common snags when implementing priority queues in your Python projects.

• Handling items with identical priorities in heapq
• When two items in a heapq-based queue share the same priority, Python attempts to compare the next element in the tuple to break the tie. If that element is a complex object or a string that doesn't support comparison, your code will raise a TypeError. The best fix is to add a unique, auto-incrementing counter as a tie-breaker, as shown in the custom class implementation earlier.
• Forgetting to implement __lt__ for custom objects in heaps
• If you push custom objects directly into a heap, you'll get a TypeError unless you tell Python how to compare them. The heapq module needs to know which object has a higher priority. You can solve this by defining the "less-than" method, __lt__, inside your custom class to specify that comparisons should be based on the priority attribute.
• Creating a max heap with heapq
• By default, heapq is a min-heap, always returning the item with the smallest value. If you need a max-heap—where the largest value has the highest priority—you can use a simple trick. Just multiply your priority numbers by -1 before pushing them onto the heap. When you pop an item, multiply its priority by -1 again to restore the original value.

Handling items with identical priorities in heapq

When items have the same priority, heapq breaks the tie by comparing the next element in the tuple. This can lead to unexpected ordering, as tasks aren't necessarily processed in a first-in-first-out sequence. The following code demonstrates this behavior.

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}")

Because the priorities are identical, heapq resorts to sorting by the task strings. This causes the tasks to be processed alphabetically, not in the order they were added. The code below shows how to correct this behavior.

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 solution introduces a counter to ensure a stable, first-in-first-out order for tasks with the same priority. By adding an incrementing number as the second element in the tuple, like (priority, counter, task), you create a unique tie-breaker. This forces heapq to process items in the order they were added when priorities are equal. It's a crucial fix that prevents both alphabetical sorting and potential TypeError issues with complex, non-comparable objects.

Forgetting to implement __lt__ for custom objects in heaps

Pushing custom objects into a heap without a comparison guide will crash your code. Python's heapq module doesn't know how to rank your objects, triggering a TypeError. The code below illustrates what happens when the required __lt__ method is missing.

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}")

This code fails because heapq tries to compare two Task objects but has no rule for it. It can't determine which object has a higher priority. The following example shows how to define this missing 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 inside your custom Task class. This special method gives heapq a clear rule for comparisons, telling it to sort objects based on their priority attribute. By defining return self.priority < other.priority, you explicitly state that a lower priority number means a higher priority in the queue. This simple addition resolves the TypeError and allows the heap to manage your custom objects correctly.

Creating a max heap with heapq

Creating a max heap with heapq

Python's heapq module is a min-heap, always returning the smallest item. This is perfect for ascending order but creates a problem when you need a max-heap to get the largest items first. The code below shows what happens.

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 output shows the numbers printed in ascending order, which is the opposite of what you'd expect from a max-heap. The next example shows how to invert this behavior with a simple trick.

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)

To simulate a max-heap with heapq, you can invert the priority values. By multiplying each number by -1 before pushing it, you trick the min-heap into treating the largest numbers as the smallest. For example, 4 becomes -4, which heapq prioritizes over -1. When you pop an item, simply multiply it by -1 again to get back the original value. This technique is essential whenever you need to process items in descending order.

Real-world applications

Beyond theory, priority queues are essential for building practical applications, from scheduling tasks with timestamps to finding the shortest path in a network.

Using heapq for task scheduling with timestamps

You can use a priority queue to manage tasks based on their execution time, ensuring that the soonest tasks are always handled first.

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 code uses heapq to manage a list of scheduled tasks. Each task is a tuple, (minutes, task), where the first element dictates its priority.

• The heapq.heappush() function adds tasks to the list, maintaining the heap property.
• The heapq.heappop() function then retrieves the item with the smallest minutes value.

This structure guarantees that items are processed based on their numerical priority, not the order they were added. It's a powerful way to handle time-sensitive operations.

Implementing Dijkstra's algorithm with heapq

The heapq module is perfect for implementing Dijkstra's algorithm, which finds the shortest path in a graph by using a priority queue to always explore the next-closest node.

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 code implements Dijkstra's algorithm to find the shortest path in a network. The priority queue, managed by heapq, is key. It always pulls the path with the lowest total cost so far, ensuring the most promising routes are explored first.

• The queue stores tuples containing the current cost, the node, and the path taken to get there.
• A visited set prevents getting stuck in loops by tracking already processed nodes.
• The loop continues until the end node is reached, guaranteeing the returned path is the most efficient one.

Get started with Replit

Now, turn your knowledge into a real tool. Describe what you want to build to Replit Agent, like “a support ticket system that prioritizes urgent issues” or “a background job processor for time-sensitive tasks.”

Replit Agent writes the code, tests for errors, and deploys your app from a simple description. Start building with Replit.