Always-Sorted, Never Slow: A Practical Deep Dive into Python's heapq

Sooner or later, most programs need to answer a deceptively simple question: what is the smallest (or largest) thing I have right now, and what's next after that? A job scheduler needs the highest-priority task. A route planner needs the closest unvisited node. A log processor needs the ten slowest requests out of millions. You could keep a list sorted and pull from the end, but re-sorting on every insert is wasteful, and a full sort is overkill when you only care about the extremes.

Python's answer lives in the standard library: heapq. It implements a binary heap — a data structure that keeps the smallest element instantly accessible while allowing pushes and pops in O(log n) time. No third-party install, no classes to learn. Just a handful of functions that operate directly on an ordinary Python list. This post walks through how it works and the patterns you'll actually reach for in real code.

The mental model: a list that's a heap

The first surprising thing about heapq is that it has no Heap object. You use a plain list, and heapq's functions maintain the heap invariant on it: every parent is smaller than or equal to its children. That makes heap[0] always the smallest element. The list won't look fully sorted — and that's fine. A heap trades total ordering for cheap access to just the minimum.

import heapq

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

# Pop them out — they come in sorted order, smallest first
print([heapq.heappop(heap) for _ in range(len(heap))])
# [1, 2, 3, 5, 8]

If you already have a list, you don't need to push items one at a time. heapify rearranges it in place in linear time:

data = [9, 4, 7, 1, 6, 3]
heapq.heapify(data)
print(data[0])   # 1  — the smallest, always at index 0

One crucial detail: heapq only gives you a min-heap. heap[0] is the minimum. We'll see how to get max-heap behavior in a moment.

Building a priority queue

The classic use of a heap is a priority queue, and the idiom is to push tuples where the first element is the priority. Tuples compare lexicographically, so the heap orders by priority automatically:

tasks = []
heapq.heappush(tasks, (2, "send email"))
heapq.heappush(tasks, (1, "deploy release"))
heapq.heappush(tasks, (3, "clean up temp files"))

print(heapq.heappop(tasks))   # (1, 'deploy release')

This works beautifully until two items share a priority and the payload isn't comparable. If both priorities are 1, Python falls back to comparing the second tuple elements — and if those are dicts or custom objects, you get a TypeError. The standard fix is to insert a monotonically increasing counter as a tie-breaker, which also gives you stable, insertion-order behavior for equal priorities:

import itertools

counter = itertools.count()
pq = []

def add_task(pq, priority, task):
    # (priority, insertion_order, payload)
    heapq.heappush(pq, (priority, next(counter), task))

add_task(pq, 1, {"name": "a"})
add_task(pq, 1, {"name": "b"})

# The counter breaks the tie; the dicts are never compared
print(heapq.heappop(pq)[2])   # {'name': 'a'}
print(heapq.heappop(pq)[2])   # {'name': 'b'}

Getting a max-heap

Since heapq is min-only, the standard trick for a max-heap is to negate the sort key. Push -value, and negate again when you pop:

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

print(-heapq.heappop(max_heap))   # 8  — the largest

For tuples you can negate just the priority field, e.g. (-priority, counter, item), leaving the payload untouched.

Top-k without sorting everything

If you only need the k largest or smallest items, don't sort the whole collection. heapq.nlargest and heapq.nsmallest do it more efficiently and accept a key function, just like sorted:

people = [
    {"name": "A", "age": 30},
    {"name": "B", "age": 25},
    {"name": "C", "age": 40},
]

oldest_two = heapq.nlargest(2, people, key=lambda p: p["age"])
print([p["name"] for p in oldest_two])   # ['C', 'A']

A rule of thumb: if k is 1, use the built-ins min/max. If k is close to the total size, a plain sorted() is clearer. For the middle ground — a small k out of a large n — nlargest/nsmallest shine.

Push-and-pop in one step

heapq offers two combined operations that are faster than doing a push and a pop separately, because they only reshuffle the heap once:

• heappushpop(heap, item) — push item, then pop and return the smallest.
• heapreplace(heap, item) — pop and return the smallest, then push item.

The difference is order, which matters when your new item might itself be the smallest:

h = [1, 3, 5]
heapq.heapify(h)

# push 4, then pop the smallest of {1,3,5,4}
print(heapq.heappushpop(h, 4))   # 1
print(h)                          # [3, 4, 5]

These are the key to an efficient streaming top-k: keep a min-heap of size k, and whenever a new value beats the current minimum, replace it. Memory stays fixed no matter how long the stream is:

def top_k(stream, k):
    heap = []
    for x in stream:
        if len(heap) < k:
            heapq.heappush(heap, x)
        elif x > heap[0]:            # heap[0] is the smallest kept so far
            heapq.heapreplace(heap, x)
    return sorted(heap, reverse=True)

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

Merging sorted streams

heapq.merge lazily combines multiple already-sorted inputs into one sorted iterator, using a heap under the hood. It's memory-friendly because it never materializes everything at once — ideal for merging sorted files or database cursors:

a = [1, 4, 7]
b = [2, 3, 8]
c = [0, 5]

print(list(heapq.merge(a, b, c)))
# [0, 1, 2, 3, 4, 5, 7, 8]

It accepts key and reverse arguments too, mirroring sorted.

A real algorithm: Dijkstra's shortest path

Heaps aren't just for scheduling — they're the engine behind many graph algorithms. Here's a compact Dijkstra using a heap as the frontier. Note the if d > dist[node]: continue line: instead of trying to update entries already in the heap (which heaps don't support cheaply), we just push new ones and skip any stale entry we pop later.

import math

def shortest_paths(graph, start):
    dist = {node: math.inf for node in graph}
    dist[start] = 0
    pq = [(0, start)]

    while pq:
        d, node = heapq.heappop(pq)
        if d > dist[node]:
            continue                     # stale entry, skip
        for neighbor, weight in graph[node].items():
            new_d = d + weight
            if new_d < dist[neighbor]:
                dist[neighbor] = new_d
                heapq.heappush(pq, (new_d, neighbor))
    return dist

graph = {
    "A": {"B": 1, "C": 4},
    "B": {"C": 2, "D": 5},
    "C": {"D": 1},
    "D": {},
}
print(shortest_paths(graph, "A"))
# {'A': 0, 'B': 1, 'C': 3, 'D': 4}

Common pitfalls

Expecting a sorted list. The underlying list is not fully sorted after heapify; only heap[0] is guaranteed to be the minimum. Iterate with repeated heappop if you need sorted output.

Comparing unorderable payloads. If you push (priority, object) tuples and two priorities tie, Python will try to compare the objects. Always add a counter or another comparable tie-breaker.

Trying to update priorities in place. A binary heap has no cheap "decrease-key." The idiomatic workaround, shown in Dijkstra above, is to push a new entry and lazily ignore outdated ones when they surface.

Thread safety. heapq functions are not synchronized. For producer/consumer work across threads, use queue.PriorityQueue, which wraps a heap with locking.

Wrap-up and next steps

heapq is a small module that punches far above its weight. Once the "it's just a list plus functions" model clicks, you'll reach for it whenever you need the next-smallest item efficiently: task schedulers, top-k over streams, merging sorted sources, and graph search. Start by replacing any "sort the whole thing just to grab a few extremes" code with nlargest/nsmallest, then graduate to full priority queues with the counter idiom.

From here, explore queue.PriorityQueue for concurrent code, and read the CPython source for heapq — it's short, well-commented pure Python, and a great way to see a binary heap implemented cleanly. Your future self, staring at a million-row log file, will thank you.