Python heapq: Build Priority Queues

When an application must always process the most urgent, smallest, or earliest item, sorting the entire collection after every change is often wasteful. Schedulers, simulations, task systems, graph algorithms, and large-data workflows usually need fast access to the next item rather than a fully sorted list. Python heapq provides a compact solution through a binary heap stored in an ordinary list.

This guide explains how to build priority queues, push and pop items, transform existing lists, select the largest or smallest values, and handle equal priorities safely. The topic connects naturally with the guides about the Python collections module, sort versus sorted, Python lists, Python functions, and algorithms in programming.

What is a heap?

A heap is a binary tree represented inside a list. In the min-heap used by default, each parent is less than or equal to its children. This rule is called the heap invariant. It does not mean that every list position is sorted. It only guarantees that the smallest element is stored at index zero.

For an index k, the child indexes are 2*k + 1 and 2*k + 2. The list representation avoids separate tree-node objects and keeps the structure compact. The official heapq documentation describes this layout, the available operations, and the APIs for minimum and maximum heaps.

Why use heapq instead of sorting everything?

Sorting n elements generally costs O(n log n). If a program inserts one task and needs only the smallest item, repeatedly sorting the whole list performs unnecessary work. A heap can insert or remove the root in O(log n), while reading the current smallest value from heap[0] costs O(1).

Converting a complete list with heapify() runs in linear time, O(n). When all initial values are already available, calling heapify once is usually better than pushing every item separately.

Turning a list into a heap

import heapq

numbers = [18, 4, 12, 7, 2, 30, 9]
heapq.heapify(numbers)

print(numbers)
print(numbers[0])  # smallest value

The resulting list may not look sorted, but it satisfies the heap invariant. Avoid testing it against a specific internal arrangement. The important contract is that index zero contains the smallest item and that heap operations preserve the structure.

Pushing and popping items

Use heappush() to insert a value and heappop() to remove the smallest one:

import heapq

queue = []
heapq.heappush(queue, 20)
heapq.heappush(queue, 5)
heapq.heappush(queue, 12)

while queue:
    next_value = heapq.heappop(queue)
    print(next_value)

The values are printed in ascending order. To inspect the current smallest value without removing it, read queue[0]. Calling heappop() on an empty heap raises IndexError, so check the queue state or handle the exception explicitly.

Building a priority queue with tuples

Real queues normally store a priority together with a task. Python compares tuples from left to right, so place the priority first:

import heapq

queue = []
heapq.heappush(queue, (3, "generate report"))
heapq.heappush(queue, (1, "restore service"))
heapq.heappush(queue, (2, "reply to customer"))

priority, task = heapq.heappop(queue)
print(priority, task)

In this convention, a smaller number means higher urgency. Another convention is possible, but the project should document it and use it consistently.

Handling equal priorities

If two priorities are equal, tuple comparison continues with the second value. Text tasks may compare successfully, but custom objects, dictionaries, or mixed values can raise TypeError. You may also want tasks with equal priority to keep their insertion order. A monotonically increasing counter solves both issues:

import heapq
from itertools import count

sequence = count()
queue = []

heapq.heappush(queue, (2, next(sequence), {"id": 101}))
heapq.heappush(queue, (2, next(sequence), {"id": 102}))

priority, order, task = heapq.heappop(queue)
print(task)

The sequence number acts as a stable tie-breaker. Since every count is unique, Python never needs to compare the task objects directly.

Using a dataclass for prioritized items

A wrapper class can make the structure clearer. Mark the payload field with compare=False so only the priority participates in ordering:

from dataclasses import dataclass, field
from typing import Any
import heapq

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    item: Any = field(compare=False)

queue = [
    PrioritizedItem(4, "backup"),
    PrioritizedItem(1, "critical alert"),
]
heapq.heapify(queue)
print(heapq.heappop(queue).item)

This pattern is useful when prioritized entries move between multiple modules or service layers.

Updating or removing pending tasks

Changing an arbitrary position can break the heap invariant. Finding a task also requires a linear scan. A common design keeps a dictionary that maps tasks to their current entries. When a priority changes, the old entry is marked as removed and a new entry is pushed. The pop operation skips stale entries.

import heapq
from itertools import count

REMOVED = object()
heap = []
entry_finder = {}
sequence = count()

def add_task(task, priority):
    if task in entry_finder:
        remove_task(task)
    entry = [priority, next(sequence), task]
    entry_finder[task] = entry
    heapq.heappush(heap, entry)

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

def pop_task():
    while heap:
        priority, _, task = heapq.heappop(heap)
        if task is not REMOVED:
            del entry_finder[task]
            return task, priority
    raise KeyError("priority queue is empty")

This lazy-deletion technique preserves efficient heap operations and avoids rebuilding the structure after every update.

heappushpop and heapreplace

The combined functions are valuable for fixed-size heaps. heappushpop(heap, item) pushes a new item and removes the smallest value in one optimized operation. It returns the smaller of the new item and the previous root, leaving the larger value in the heap.

heapreplace(heap, item) removes the existing root first and then inserts the new value. The heap size remains unchanged, but the heap must not be empty. The distinction matters when maintaining the three largest values observed so far:

import heapq

largest = [8, 2, 15]
heapq.heapify(largest)

for value in [3, 21, 7]:
    if value > largest[0]:
        heapq.heapreplace(largest, value)

print(sorted(largest, reverse=True))

The heap keeps only three values. Its root is the smallest current candidate and can be replaced whenever a better value arrives.

Finding the largest and smallest values

nlargest() and nsmallest() return a limited number of elements and support a key function:

import heapq

products = [
    {"name": "A", "price": 80},
    {"name": "B", "price": 25},
    {"name": "C", "price": 110},
    {"name": "D", "price": 45},
]

most_expensive = heapq.nlargest(2, products, key=lambda p: p["price"])
cheapest = heapq.nsmallest(1, products, key=lambda p: p["price"])
print(most_expensive)
print(cheapest)

These functions work best when n is small relative to the dataset. If most elements are needed, sorted() may be faster and clearer. For exactly one result, use the built-in min() or max().

Maximum heaps in Python 3.14

For many years, developers simulated a max-heap by storing negative numbers. Python 3.14 introduced explicit maximum-heap functions: heapify_max(), heappush_max(), heappop_max(), heappushpop_max(), and heapreplace_max().

import heapq

values = [4, 19, 7, 12]
heapq.heapify_max(values)
print(heapq.heappop_max(values))  # 19

Projects supporting older Python releases can still negate numeric priorities, but the convention must be documented carefully. The newer _max APIs express intent directly and avoid sign mistakes.

Merging sorted streams

heapq.merge() combines multiple already-sorted inputs and returns an iterator. Unlike concatenating every value and calling sorted, it can process streams without loading all records into memory:

import heapq

log_a = [1, 4, 8]
log_b = [2, 3, 10]

for value in heapq.merge(log_a, log_b):
    print(value)

This is useful for timestamped logs, sorted files, paginated exports, and other ordered data sources. Every input must use the same ordering direction.

heapq or queue.PriorityQueue?

heapq is a set of functions operating on a list. It does not provide synchronization or blocking behavior. When multiple threads produce and consume work, the class documented in queue.PriorityQueue offers locking, optional capacity limits, and waiting operations.

Choose heapq for local algorithms, single-threaded work, or code that already controls synchronization. Choose PriorityQueue for a ready-made thread-safe queue. Asynchronous programs can also use asyncio.PriorityQueue.

A small task scheduler example

A scheduler can order jobs by execution time. The earliest timestamp is always removed first:

import heapq
from datetime import datetime, timedelta

schedule = []
now = datetime.now()

heapq.heappush(schedule, (now + timedelta(minutes=10), "send report"))
heapq.heappush(schedule, (now + timedelta(minutes=2), "refresh cache"))
heapq.heappush(schedule, (now + timedelta(minutes=5), "check imports"))

when, task = heapq.heappop(schedule)
print(when, task)

A production scheduler also needs persistence, time-zone rules, retries, cancellation, and protection against duplicate workers. The heap solves the ordering problem, not the entire job-processing lifecycle.

Common mistakes

• Assuming the complete list is sorted because heap[0] is the smallest value.
• Using append() instead of heappush() after the list becomes a heap.
• Removing arbitrary indexes and breaking the invariant.
• Comparing incompatible tasks when equal priorities occur.
• Using negative values for a max-heap without documenting the convention.
• Calling nlargest for almost every element when sorting would be simpler.
• Sharing the list across threads without synchronization.

Testing a priority queue

Do not assert one exact internal list shape, because several arrangements can satisfy the heap invariant. Test observable behavior: pop order, tie handling, updates, lazy removal, and empty-queue errors.

def test_priority_order():
    queue = []
    heapq.heappush(queue, (3, "low"))
    heapq.heappush(queue, (1, "high"))
    heapq.heappush(queue, (2, "medium"))

    assert heapq.heappop(queue)[1] == "high"
    assert heapq.heappop(queue)[1] == "medium"
    assert heapq.heappop(queue)[1] == "low"

Best practices

• Define whether a smaller number means a higher priority.
• Use a sequence counter to preserve insertion order.
• Wrap the heap in a class when updates and cancellation are required.
• Use heapify() for an existing collection.
• Prefer combined operations for bounded heaps.
• Measure before replacing a simple sort.
• Document the minimum Python version when using max-heap APIs.

Conclusion

Python heapq provides fast access to priority items without sorting the entire collection after every modification. With heapify, heappush, heappop, combined operations, selection helpers, stream merging, and maximum-heap APIs, it supports both compact algorithms and practical task queues.

The key is to preserve the heap invariant and design the entry format carefully. A simple priority-and-task tuple is enough for basic cases. Equal priorities, updates, and cancellation require a sequence counter, an auxiliary dictionary, and lazy deletion. With those patterns, the queue remains efficient, predictable, and straightforward to test.