Priority Queues Done Right: A Practical Deep Dive into Python's heapq

Sooner or later every program needs to answer the question: of everything I'm holding right now, what should I handle next? The task scheduler wants the highest-priority job. The pathfinder wants the closest unexplored node. The analytics job wants the ten largest values out of a billion. Sorting the whole collection every time you need the front item is wasteful, and keeping a list sorted on every insert is O(n) per element. What you want is a priority queue, and Python ships one in the standard library: heapq.

The catch is that heapq doesn't give you a tidy PriorityQueue class. It gives you a handful of functions that operate on an ordinary list, treating that list as a binary heap. That design feels strange at first, but once it clicks you'll reach for it constantly. This post walks through the whole module with runnable examples, then builds three things people actually use it for: top-N queries, a streaming median, and Dijkstra's shortest path.

What a heap actually is

A binary heap is a complete binary tree where every parent is smaller than or equal to its children. Python's heapq implements a min-heap, so the smallest element is always at the root. The clever part is that the tree is stored flat in a list: the children of index i live at 2*i + 1 and 2*i + 2. That means heap[0] is always the minimum, and both insertion and removal cost O(log n) because they only touch one path from root to leaf.

You start with a plain list and use the module's functions to maintain the heap invariant:

import heapq

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

print("smallest:", heap[0])          # smallest: 1

order = [heapq.heappop(heap) for _ in range(len(heap))]
print("sorted order:", order)        # [1, 2, 3, 5, 8, 9]

Two things to notice. First, heap[0] peeks at the minimum in constant time without removing it. Second, repeatedly popping yields elements in sorted order — that's the basis of heapsort. But the heap itself is not a sorted list; if you print it mid-way you'll see something like [1, 3, 2, 5, 9, 8]. It's only guaranteed that each parent is <= its children, which is exactly enough to keep the smallest item at the front.

heapify: turning a list into a heap in place

If you already have all your data, don't push items one at a time. heapq.heapify() rearranges an existing list into heap order in a single O(n) pass — faster than n separate O(log n) pushes.

data = [5, 1, 8, 3, 9, 2]
heapq.heapify(data)          # in place, returns None
print(data)                  # [1, 3, 2, 5, 9, 8]
print("min:", data[0])       # min: 1

A common gotcha: heapify mutates the list and returns None, so data = heapq.heapify(data) would throw your data away. Call it as a statement.

The efficient combos: heappushpop and heapreplace

When you want to add something and immediately take the smallest, you could call heappush then heappop — but that reorganizes the heap twice. The module gives you two fused operations that do it in a single sift:

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

# push 0, then pop the smallest -> returns 0 (the item you just added)
print(heapq.heappushpop(h, 0))   # 0

# push 4, then pop the smallest -> returns 1
print(heapq.heappushpop(h, 4))   # 1

# pop the smallest FIRST, then push 6 -> returns the old min
print(heapq.heapreplace(h, 6))   # 3

The difference is the order of operations. heappushpop adds first, so if your new item is the smallest it comes straight back out. heapreplace pops first, so it always returns an item that was already on the heap (and it errors on an empty heap). These are the workhorses for "keep only the N best" patterns, shown below.

Making a max-heap

heapq is min-only, but a max-heap is one negation away. Store the negatives, and the smallest negative corresponds to the largest original value:

nums = [5, 1, 8, 3, 9, 2]
maxheap = [-n for n in nums]
heapq.heapify(maxheap)

largest = -heapq.heappop(maxheap)
print("largest:", largest)       # largest: 9

For numbers this is clean. For objects, wrap them in a (-priority, item) tuple instead, which leads directly to the next topic.

Priorities and the tie-breaking trap

The idiomatic way to build a real priority queue is to push (priority, item) tuples. Tuples compare lexicographically, so the heap orders by priority automatically. But there's a subtle bug waiting: if two items share a priority, Python falls back to comparing the second tuple element. If those are dicts, custom objects, or anything without a natural ordering, you get a TypeError mid-run.

The standard fix is to insert a monotonically increasing counter as a tie-breaker. It's always unique, so comparison never reaches the payload, and it preserves insertion order (FIFO) among equal priorities:

import heapq, itertools

counter = itertools.count()
pq = []

for priority, task in [(2, "email"), (1, "deploy"), (2, "lunch"), (1, "hotfix")]:
    heapq.heappush(pq, (priority, next(counter), task))

processed = []
while pq:
    priority, _, task = heapq.heappop(pq)
    processed.append(task)

print(processed)   # ['deploy', 'hotfix', 'email', 'lunch']

Both priority-1 tasks come out before the priority-2 tasks, and within each priority they preserve the order they were added. Remember lower numbers are "more urgent" here because it's a min-heap; negate the priority if you want larger numbers to win.

Top-N without sorting everything

When you only need the largest or smallest handful from a big iterable, nlargest and nsmallest beat sorted(...)[:n] because they keep a bounded heap of size n instead of sorting the entire input. Both accept a key function, just like sorted:

people = [
    {"name": "Ann", "age": 34},
    {"name": "Bob", "age": 21},
    {"name": "Cid", "age": 55},
    {"name": "Dot", "age": 43},
]

oldest = heapq.nlargest(2, people, key=lambda p: p["age"])
print([p["name"] for p in oldest])   # ['Cid', 'Dot']

A rule of thumb: if n is 1, plain min()/max() is faster. If n is close to the total size, just sorted(). But for the common case of "top 10 out of millions," nlargest/nsmallest are the right tool and run in roughly O(m log n) memory-bounded time.

Merging sorted streams

heapq.merge() lazily merges multiple already-sorted inputs into one sorted stream. It's a generator, so it never materializes everything at once — ideal for merging sorted files or database cursors too large for memory:

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

for x in heapq.merge(a, b, c):
    print(x, end=" ")    # 0 1 2 3 4 5 7 8 9

It also takes key and reverse arguments (matching the sort order of your inputs). The important word is sorted: merge assumes each input is already ordered and only interleaves them — it will not sort unsorted data for you.

A streaming median with two heaps

Here's a classic that shows heaps working together. To track the median of a growing stream of numbers, keep two heaps: a max-heap of the lower half and a min-heap of the upper half. Balance their sizes and the median is always at one or both tops — no re-sorting as data arrives.

import heapq

class MedianStream:
    def __init__(self):
        self.low = []   # max-heap (store negatives)
        self.high = []  # min-heap

    def add(self, x):
        heapq.heappush(self.low, -x)
        # move the low half's max into the high half
        heapq.heappush(self.high, -heapq.heappop(self.low))
        # rebalance so low is never smaller than high
        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

m = MedianStream()
for x in [5, 2, 8, 1, 9, 3]:
    m.add(x)
print(m.median())   # 4.0

Each add is O(log n) and median is O(1). Try feeding it an odd-length stream like [5, 2, 8, 1, 9] and you'll get 5 — the middle value — because low ends up holding one extra element.

Dijkstra's shortest path

Priority queues are the beating heart of Dijkstra's algorithm: always expand the closest unexplored node next. With heapq it's remarkably compact. The one trick worth knowing is the "lazy deletion" check — instead of updating entries already on the heap, we push new ones and skip any stale entry whose recorded distance is worse than the best we've since found:

import heapq

def dijkstra(graph, start):
    dist = {start: 0}
    pq = [(0, start)]
    while pq:
        d, node = heapq.heappop(pq)
        if d > dist.get(node, float("inf")):
            continue                      # stale entry, skip
        for nbr, weight in graph[node]:
            nd = d + weight
            if nd < dist.get(nbr, float("inf")):
                dist[nbr] = nd
                heapq.heappush(pq, (nd, nbr))
    return dist

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

The shortest route to D is A→B→C→D with cost 4, not the direct A→C→D that a greedy first guess might take. The heap guarantees we finalize nodes in order of increasing distance.

Common pitfalls

A few things trip people up. The heap is not a sorted list — never rely on the order of elements past index 0. heapify and the push/pop functions mutate in place and mostly return None (except the pop-style ones), so don't reassign their results. Always add a tie-breaker when pushing tuples whose payloads aren't comparable. And if you find yourself needing thread-safe blocking behavior across producers and consumers, reach for queue.PriorityQueue instead — it wraps heapq with locking, at some overhead cost.

Wrap-up and next steps

You now have the whole toolbox: heappush/heappop for one-at-a-time work, heapify for bulk loading, heappushpop/heapreplace for fused updates, nlargest/nsmallest for top-N, and merge for sorted streams — plus the tuple-with-counter pattern that turns any of them into a proper priority queue. From here, try building a bounded "keep the N best" filter using heappushpop on a fixed-size heap, or extend the Dijkstra sketch to reconstruct the actual path by tracking predecessors. Once heaps are in your reflexes, a whole category of "what's next?" problems stops being hard.