Heap and heapq in Python: Priority Queues

A heap is a structure that always gives you the smallest item first, quickly, no matter in which order things were added. It is the engine behind a priority queue: a queue where the item with the highest priority leaves first, not the one that waited longest. Task schedulers, “top 10” lists, shortest-path algorithms and merging sorted data all use one.

Python’s standard library has a ready-made heap in the heapq module. This guide explains how it works, how to use it (including the max-heap trick and the tie-breaking trap), and the classic problems it solves: top-k, the median of a stream, and Dijkstra’s shortest path. All the output shown is real.

What is a heap?

A (min-)heap is a binary tree with one rule: every parent is smaller than or equal to its children. So the smallest item is always at the top. Unlike a binary search tree, there is no order between the left and the right child. The tree is also always packed: every level is full, except possibly the last, which fills from the left.

Because it is packed, no node objects are needed. The tree is stored in a plain list, level by level:

list:   [1, 2, 8, 5, 3]

tree:        1
           /   \
          2     8
         / \
        5   3

For the item at index i, the children are at 2i + 1 and 2i + 2, and the parent is at (i - 1) // 2:

heap = [1, 2, 8, 5, 3]
for i in range(len(heap)):
    left, right = 2 * i + 1, 2 * i + 2
    kids = [heap[k] for k in (left, right) if k < len(heap)]
    print(f"index {i} value {heap[i]}: children {kids}")

Output

index 0 value 1: children [2, 8]
index 1 value 2: children [5, 3]
index 2 value 8: children []
index 3 value 5: children []
index 4 value 3: children []

The item 2 at index 1 has children 5 and 3 at indexes 3 and 4. When you add an item, it is placed at the end and swaps upward while it is smaller than its parent. When you remove the smallest, the last item takes the root’s place and swaps downward. A tree with a million items is only about 20 levels tall, so both take about 20 steps at most. That is O(log n).

The heapq module

The functions work on an ordinary list. Push items with heappush and take the smallest with heappop. The smallest can be read without removing it from heap[0]. Look at the printed list: it is a valid heap, but it is not sorted. Only the first item is guaranteed:

import heapq

heap = []
for x in [5, 3, 8, 1, 2]:
    heapq.heappush(heap, x)

print(heap)                    # a valid heap, NOT a sorted list
print(heap[0])                 # the smallest item is always first
print(heapq.heappop(heap), heapq.heappop(heap), heapq.heappop(heap))
print(heap)

Output

[1, 2, 8, 5, 3]
1
1 2 3
[5, 8]

If you already have a list, heapify rearranges it into a heap in place in O(n), which is faster than pushing the items one at a time. Notice that the result differs from the pushes above. Many different lists are valid heaps of the same items:

import heapq

nums = [5, 3, 8, 1, 2]
heapq.heapify(nums)            # rearranges the list in place, O(n)
print(nums)
print(nums[0])
print(nums == sorted(nums))

Output

[1, 2, 8, 3, 5]
1
False

Since a heap is not fully sorted, do not loop over it or slice it and expect ordered results. To get the items in order, pop them one by one. That is heap sort: n pops of O(log n) each make O(n log n) in total:

import heapq

def heap_sort(items):
    heap = list(items)
    heapq.heapify(heap)
    return [heapq.heappop(heap) for _ in range(len(heap))]

print(heap_sort([5, 2, 9, 1, 5, 6]))

Output

[1, 2, 5, 5, 6, 9]

For sorting in real code, use sorted() (see Sorting Algorithms in Python). A heap is for when the data keeps arriving and you keep asking for the smallest.

Max-heap and priorities

heapq only provides a min-heap. To get the largest item first, store each number negated, and negate it again when you take it out:

import heapq

heap = []
for x in [5, 3, 8, 1]:
    heapq.heappush(heap, -x)          # store negatives
print(-heapq.heappop(heap), -heapq.heappop(heap))

Output

8 5

For a priority queue, push tuples of (priority, item). Python compares tuples left to right, so the lowest priority number comes out first, and equal priorities are decided by the next element:

import heapq

tasks = []
heapq.heappush(tasks, (2, "write report"))
heapq.heappush(tasks, (1, "fix bug"))
heapq.heappush(tasks, (3, "lunch"))
heapq.heappush(tasks, (1, "deploy"))

while tasks:
    priority, name = heapq.heappop(tasks)
    print(priority, name)

Output

1 deploy
1 fix bug
2 write report
3 lunch

The “deploy” task came before “fix bug” only because the string "deploy" is alphabetically smaller. That works for strings, but for anything that cannot be compared, it fails. Two jobs with the same priority and dictionaries as payload raise a TypeError. The usual fix is a running counter as the second element, so the payload is never compared and equal priorities come out in insertion order:

import heapq, itertools

heap = []
heapq.heappush(heap, (1, {"job": "a"}))
try:
    heapq.heappush(heap, (1, {"job": "b"}))   # same priority: Python compares the dicts
except TypeError as e:
    print(type(e).__name__)

# the fix: a counter that breaks ties, so payloads are never compared
heap, counter = [], itertools.count()
for job in ["a", "b", "c"]:
    heapq.heappush(heap, (1, next(counter), {"job": job}))
print([heapq.heappop(heap)[2]["job"] for _ in range(3)])

Output

TypeError
['a', 'b', 'c']

The standard library also has queue.PriorityQueue, which wraps a heap for use between threads. For everything else, use heapq directly. (For the plain first-in-first-out queue, see Queue and Deque in Python Explained.)

nlargest, nsmallest and merge

Common needs have their own functions. nlargest(k, items) and nsmallest(k, items) return the k best without sorting everything, and both accept a key. heapq.merge merges several already sorted inputs into one sorted stream, lazily, without loading them all into memory:

import heapq

scores = [72, 91, 65, 88, 95, 70]
print(heapq.nlargest(3, scores))
print(heapq.nsmallest(2, scores))

people = [("Asha", 30), ("Ben", 25), ("Chen", 41)]
print(heapq.nlargest(1, people, key=lambda p: p[1]))

Output

[95, 91, 88]
[65, 70]
[('Chen', 41)]

import heapq

print(list(heapq.merge([1, 4, 7], [2, 5, 8], [3, 6, 9])))
print(list(heapq.merge([5, 3, 1], [6, 4, 2], reverse=True)))

Output

[1, 2, 3, 4, 5, 6, 7, 8, 9]
[6, 5, 4, 3, 2, 1]

For a large k, close to the size of the list, plain sorted(items)[:k] is often quicker. And for a single smallest or largest value, min() and max() are all you need.

Top-k without sorting everything

“Find the 3 biggest numbers in a huge stream.” Sorting everything costs O(n log n) and needs all data in memory. Instead, keep a min-heap of the k biggest seen so far. Its top is the smallest of them, which is the bar a new number has to beat. If a new number is bigger, it replaces the top:

import heapq

def top_k(stream, k):
    heap = []                                  # a min-heap holding the k biggest so far
    for x in stream:
        if len(heap) < k:
            heapq.heappush(heap, x)
        elif x > heap[0]:                      # bigger than the smallest of the top k
            heapq.heapreplace(heap, x)         # pop the smallest and push x in one step
    return sorted(heap, reverse=True)

print(top_k([5, 1, 9, 3, 7, 8, 2], 3))
print(top_k([4, 4], 5))

Output

[9, 8, 7]
[4, 4]

Each number costs at most O(log k), and memory stays at k items however long the stream is. It is a good example of choosing the “opposite” heap: to find the largest k, use a min-heap.

Median of a stream (two heaps)

A famous interview problem: numbers arrive one at a time, and you must report the median after each. Keep the smaller half in a max-heap and the larger half in a min-heap, and keep the two halves the same size (or the lower one bigger by one). The median is then at the tops of the heaps, in O(1):

import heapq

class MedianFinder:
    def __init__(self):
        self.low = []       # max-heap (negated) for the smaller half
        self.high = []      # min-heap for the larger half

    def add(self, x):
        heapq.heappush(self.low, -x)
        heapq.heappush(self.high, -heapq.heappop(self.low))   # biggest of the low half moves up
        if len(self.high) > len(self.low):
            heapq.heappush(self.low, -heapq.heappop(self.high))

    def median(self):
        if len(self.low) > len(self.high):
            return -self.low[0]
        return (-self.low[0] + self.high[0]) / 2

mf = MedianFinder()
medians = []
for x in [5, 2, 9, 1]:
    mf.add(x)
    medians.append(mf.median())
print(medians)

Output

[5, 3.5, 5, 3.5]

After 5, 2, 9 the median is 5, and after 1 is added the median of 1, 2, 5, 9 is (2 + 5) / 2 = 3.5. Each insert is O(log n).

Shortest paths with Dijkstra

BFS finds shortest paths when every edge counts as one step (see Graphs in Python: Adjacency List, BFS and DFS). When edges have different weights, use Dijkstra’s algorithm, which is BFS with a priority queue: always expand the node that is currently closest to the start. The heap holds (distance, node) pairs. Entries that are out of date are skipped when they come out:

import heapq

def dijkstra(graph, start):
    dist = {start: 0}
    heap = [(0, start)]
    while heap:
        d, node = heapq.heappop(heap)
        if d > dist.get(node, float("inf")):
            continue                               # an old, longer entry: skip it
        for nb, weight in graph[node]:
            new_d = d + weight
            if new_d < dist.get(nb, float("inf")):
                dist[nb] = new_d
                heapq.heappush(heap, (new_d, nb))
    return dist

graph = {
    "A": [("B", 4), ("C", 1)],
    "B": [("A", 4), ("C", 2), ("D", 1)],
    "C": [("A", 1), ("B", 2), ("D", 5)],
    "D": [("B", 1), ("C", 5)],
}
print(dict(sorted(dijkstra(graph, "A").items())))

Output

{'A': 0, 'B': 3, 'C': 1, 'D': 4}

The direct road A to B costs 4, but going A to C to B costs 1 + 2 = 3, and A to D is 4 through C and B. Dijkstra needs edges that are not negative. The time is O((V + E) log V).

Complexity

Compare this with keeping a sorted list. Finding the smallest is instant there too, but every insert costs O(n), because items shift. A heap keeps both insert and remove at O(log n). (See Time Complexity of Python Data Structures.)

Common mistakes

• Expecting the list to be sorted. Only heap[0] is guaranteed. Iterating or printing a heap shows an arrangement that is not in order.
• Expecting a max-heap. heapq is min-only. Negate numbers or wrap the priority.
• Changing the list by hand. Using append, sort or item assignment on a heap breaks the rule, unless you call heapify afterwards.
• Ties between uncomparable payloads. Add a counter as a tie-breaker, or the second push can raise TypeError.
• Popping an empty heap. heappop([]) raises IndexError. Check if heap: first.
• Re-sorting after every insert. That is O(n log n) per insert. A heap does it in O(log n).

Try it yourself

Work out each answer first, then open the solution.

1. Find the 3 largest and the smallest number in [14, 3, 27, 8, 19, 5] using heapq.

import heapq

nums = [14, 3, 27, 8, 19, 5]
print(heapq.nlargest(3, nums))
print(heapq.nsmallest(1, nums))

[27, 19, 14]
[3]

2. Design KthLargest(k, nums) with an add(x) method that returns the kth largest number seen so far.

import heapq

class KthLargest:
    def __init__(self, k, nums):
        self.k = k
        self.heap = []
        for x in nums:
            self.add(x)

    def add(self, x):
        if len(self.heap) < self.k:
            heapq.heappush(self.heap, x)
        elif x > self.heap[0]:
            heapq.heapreplace(self.heap, x)
        return self.heap[0]

kl = KthLargest(3, [4, 5, 8, 2])
print([kl.add(x) for x in [3, 5, 10, 9, 4]])

[4, 5, 5, 8, 8]

3. Stones have weights. Repeatedly smash the two heaviest together: if they are equal both vanish, otherwise the difference stays. What weight is left at the end of [2, 7, 4, 1, 8, 1]?

import heapq

def last_stone(stones):
    heap = [-s for s in stones]
    heapq.heapify(heap)
    while len(heap) > 1:
        a = -heapq.heappop(heap)          # heaviest
        b = -heapq.heappop(heap)          # second heaviest
        if a != b:
            heapq.heappush(heap, -(a - b))
    return -heap[0] if heap else 0

print(last_stone([2, 7, 4, 1, 8, 1]))

1

4. Why is heapify on a whole list faster than pushing the items one by one?

Run these in our free Python compiler.

Frequently asked questions

What is a heap in Python?

A heap is a tree-shaped structure stored in a list in which every parent is smaller than its children, so the smallest item is always first. Python provides it through the heapq module.

How do you create a max-heap in Python?

heapq only supports a min-heap. To simulate a max-heap, push the negative of each value and negate it again when you pop.

What is the time complexity of heapq operations?

heappush and heappop are O(log n), reading heap[0] is O(1), and heapify is O(n).

What is the difference between a heap and a priority queue?

A priority queue is the abstract idea: the item with the highest priority comes out first. A heap is the usual way of implementing it efficiently.

Why is my heap not sorted?

A heap only guarantees that the smallest item is at index 0 and that each parent is smaller than its children. The rest of the list is in no particular order. Pop items one by one to read them in order.

How do I find the k largest elements in Python?

Use heapq.nlargest(k, items). For a stream of data, keep a min-heap of size k and replace its top whenever a bigger number arrives.

Related reading

• Graphs in Python: Adjacency List, BFS and DFS – the graphs Dijkstra runs on.
• Queue and Deque in Python Explained – the first-in-first-out queue.
• Sorting Algorithms in Python: Bubble to Merge – heap sort compared with the others.
• Time Complexity of Python Data Structures – the cost of every operation.