How to Use Priority Queues with Queue and Heapdict in Python

A priority queue is a special data structure where each element has an associated priority. Elements with higher priority are dequeued before those with lower priority, and if two elements share the same priority, they follow their original insertion order (FIFO). Priority queues are essential in algorithms like Dijkstra's shortest path, A* search, task scheduling, and event-driven simulations.

Python provides several ways to implement priority queues. In this guide, you will learn how to use two specific approaches: the built-in queue.PriorityQueue class and the third-party heapdict module. Each serves different use cases: PriorityQueue excels in thread-safe environments, while heapdict shines when you need to update priorities dynamically.

Using queue.PriorityQueue

The queue.PriorityQueue class is part of Python's standard library and provides a thread-safe priority queue implementation. Items are stored internally using a min-heap, meaning the item with the lowest priority number is retrieved first.

Key Methods

Method	Description
put(item)	Adds an item to the queue
get()	Removes and returns the lowest-priority item
qsize()	Returns the current number of items in the queue
empty()	Returns True if the queue is empty
full()	Returns True if the queue has reached its maxsize

caution

The methods empty(), full(), and qsize() are not reliable in multi-threaded programs. Between the time you check the queue's state and the time you act on it, another thread may have modified the queue (a race condition).

Basic Example

Items are added as tuples where the first element is the priority (lower number = higher priority) and the second element is the data:

from queue import PriorityQueue

pq = PriorityQueue()

pq.put((2, "Write tests"))
pq.put((3, "Refactor code"))
pq.put((4, "Update docs"))
pq.put((5, "Clean up logs"))
pq.put((1, "Fix critical bug"))

print(pq.get())
print(pq.get())

print("Items remaining:", pq.qsize())
print("Is queue empty:", pq.empty())
print("Is queue full:", pq.full())

Output:

(1, 'Fix critical bug')
(2, 'Write tests')
Items remaining: 3
Is queue empty: False
Is queue full: False

The item with priority 1 ("Fix critical bug") is dequeued first, followed by the item with priority 2.

Setting a Maximum Size

You can limit the queue's capacity by passing a maxsize argument. When the queue is full, put() will block until space becomes available:

from queue import PriorityQueue

pq = PriorityQueue(maxsize=3)

pq.put((1, "Task A"))
pq.put((2, "Task B"))
pq.put((3, "Task C"))

print("Is queue full:", pq.full())  # True
print("Queue size:", pq.qsize())    # 3

Output:

Is queue full: True
Queue size: 3

Handling Same-Priority Items

When multiple items share the same priority, Python compares the next element in the tuple. If that element is not comparable, a TypeError is raised.

Common mistake

from queue import PriorityQueue

pq = PriorityQueue()
pq.put((1, {"task": "A"}))
pq.put((1, {"task": "B"}))  # Same priority, dicts are not comparable

Output:

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

The fix

Use a tie-breaking counter to preserve insertion order.

from queue import PriorityQueue

pq = PriorityQueue()
counter = 0

for priority, task in [(1, {"task": "A"}), (1, {"task": "B"}), (2, {"task": "C"})]:
    pq.put((priority, counter, task))
    counter += 1

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

Output:

Priority 1: {'task': 'A'}
Priority 1: {'task': 'B'}
Priority 2: {'task': 'C'}

tip

The counter ensures that items with the same priority are dequeued in the order they were inserted, and Python never needs to compare the dictionary objects directly.

Using heapdict

The heapdict module provides a dictionary-based priority queue that combines the interface of a Python dictionary with the behavior of a min-heap. Its standout feature is the ability to efficiently update priorities for existing keys, which is something that queue.PriorityQueue and heapq do not support natively.

This makes heapdict particularly useful for graph algorithms like Dijkstra's shortest path and A* search, where node priorities (distances) are frequently updated.

Installation

pip install heapdict

Key Methods

Method	Description
hd[key] = priority	Adds or updates the priority of a key
peekitem()	Returns (key, priority) with the lowest priority without removing it
popitem()	Removes and returns (key, priority) with the lowest priority
get(key, default)	Returns the priority for a key, or default if not found
keys()	Returns a view of all keys
values()	Returns a view of all priorities
items()	Returns a view of all (key, priority) pairs
clear()	Removes all items

Basic Example

import heapdict

hd = heapdict.heapdict()

hd["Write tests"] = 2
hd["Fix critical bug"] = 1
hd["Refactor code"] = 3
hd["Update docs"] = 4

print("All items:", list(hd.items()))
print("Peek lowest priority:", hd.peekitem())
print("Pop lowest priority:", hd.popitem())
print("Remaining items:", list(hd.items()))

Output:

All items: [('Write tests', 2), ('Fix critical bug', 1), ('Refactor code', 3), ('Update docs', 4)]
Peek lowest priority: ('Fix critical bug', 1)
Pop lowest priority: ('Fix critical bug', 1)
Remaining items: [('Write tests', 2), ('Refactor code', 3), ('Update docs', 4)]

Updating Priorities Dynamically

This is where heapdict truly shines. You can change the priority of an existing key, and the heap automatically re-adjusts:

import heapdict

hd = heapdict.heapdict()

hd["Node A"] = 10
hd["Node B"] = 5
hd["Node C"] = 8

print("Before update:")
print("  Lowest priority:", hd.peekitem())

# Discovered a shorter path to Node A
hd["Node A"] = 2

print("After updating 'Node A' priority to 2:")
print("  Lowest priority:", hd.peekitem())

Output:

Before update:
  Lowest priority: ('Node B', 5)
After updating 'Node A' priority to 2:
  Lowest priority: ('Node A', 2)

info

With heapq or queue.PriorityQueue, updating priorities requires workarounds like lazy deletion (marking old entries as invalid). heapdict handles this natively in O(log n) time.

Using get() for Safe Key Lookup

The get() method lets you safely look up a key's priority without risking a KeyError:

import heapdict

hd = heapdict.heapdict()
hd["server-1"] = 3
hd["server-2"] = 1

print("server-2 priority:", hd.get("server-2"))
print("server-99 priority:", hd.get("server-99", "Not found"))

Output:

server-2 priority: 1
server-99 priority: Not found

Clearing the Queue

import heapdict

hd = heapdict.heapdict()
hd["x"] = 1
hd["y"] = 2

print("Before clear:", list(hd.items()))
hd.clear()
print("After clear:", list(hd.items()))

Output:

Before clear: [('x', 1), ('y', 2)]
After clear: []

queue.PriorityQueue vs. heapdict: When to Use Which

Feature	queue.PriorityQueue	heapdict
Built-in	✅ Yes (standard library)	❌ No (requires pip install)
Thread-safe	✅ Yes	❌ No
Update priorities	❌ Not natively supported	✅ Yes (O(log n))
Dict-like access	❌ No	✅ Yes
Best for	Multi-threaded task queues	Graph algorithms (Dijkstra, A*)
Peek without removing	❌ No built-in method	✅ peekitem()

Practical Example: Simple Task Scheduler

Here is a complete example combining both approaches to build a simple task scheduler:

from queue import PriorityQueue

def run_scheduler(tasks):
    pq = PriorityQueue()
    counter = 0

    for priority, task_name in tasks:
        pq.put((priority, counter, task_name))
        counter += 1

    print("Executing tasks in priority order:")
    while not pq.empty():
        priority, _, task_name = pq.get()
        print(f"  [{priority}] {task_name}")

tasks = [
    (3, "Send email notifications"),
    (1, "Process payment"),
    (2, "Generate invoice"),
    (1, "Validate user input"),
]

run_scheduler(tasks)

Output:

Executing tasks in priority order:
  [1] Process payment
  [1] Validate user input
  [2] Generate invoice
  [3] Send email notifications

Summary

Python provides flexible tools for implementing priority queues:

• Use queue.PriorityQueue when you need a thread-safe priority queue from the standard library. It is ideal for multi-threaded applications and producer-consumer patterns.
• Use heapdict when you need to update priorities dynamically for existing keys. It is the best choice for graph algorithms like Dijkstra's shortest path and A* search where priorities change frequently.

Both implementations use a min-heap internally, meaning the item with the lowest priority number is always dequeued first. Choose the one that best matches your application's requirements for thread safety, priority updates, and API style.