Heaps and Priority Queues in Python: Always Knowing the Smallest Item

A heap answers one question, and it answers it instantly: what is the smallest item in this collection right now? Not the second smallest, not where a particular value is, not the items in order. Just the minimum, available at index 0, every time you look.

That narrowness is the point. Because a heap promises so little, it can keep its promise cheaply: adding costs O(log n), removing the smallest costs O(log n), and turning an arbitrary list of n items into a heap costs O(n) — genuinely linear, and the proof of that is the most interesting thing here. All of it happens inside a plain list, with no node objects and no pointers, because the tree structure is encoded in the array indices themselves.

You have already used one. Python's sched module keeps its pending events in a heap, asyncio's event loop keeps every scheduled timer in one, and collections.Counter.most_common(3) is a heap query. Dijkstra's algorithm and Huffman coding are both mostly a loop around a priority queue.

The idea

Two rules, and that is all

A binary min-heap is a binary tree obeying exactly two rules:

1. Shape. Every level is completely full except possibly the last, which fills left to right with no gaps. Such a tree is called complete.
2. Order. Every parent is no larger than either of its children.

Rule 2 is far weaker than being sorted. It says nothing about left child versus right child, and nothing about cousins. The one thing it buys is that the smallest item sits at the root: every node is beaten by its parent, and the chain of parents ends there.

Read that tree level by level and you get [2, 5, 3, 9, 7, 4], nowhere near sorted — 9 sits under 5 while 4 sits under 3, and the heap has no opinion about which is bigger. That weakness is exactly why a heap is cheap to maintain and a sorted list is not.

The tree lives in a flat array

Number the nodes level by level from 0. Rule 1 forbids gaps, so the numbering has no holes either, and the parent-child relationships collapse into arithmetic:

• the left child of index i is at 2 * i + 1
• the right child of index i is at 2 * i + 2
• the parent of index i is at (i - 1) // 2

The floor division makes one parent formula serve both children. A left child at 2 * i + 1 loses its one and halves back to i; a right child at 2 * i + 2 becomes the odd 2 * i + 1, and the division throws the remainder away to land on i again.

So there is no tree object anywhere. A heap of a million numbers is one list of a million slots — no node instances, no left and right references, no per-element allocation. Compare a binary search tree, where every value costs a separate object holding two pointers.

One more consequence of the shape rule: index i has a left child only while 2 * i + 1 is inside the array, which fails from n // 2 onwards. The whole second half of the list is leaves — a fact the complexity proof leans on hard.

Two repairs, each walking one path

Sift up repairs an addition. Append the new value at the end — the only position the shape rule permits — then compare it with its parent and swap while it is smaller. Stop at the root, or as soon as the parent is no larger.

Sift down repairs a removal. The root has gone, so refill it and push the replacement downwards: compare it with the smaller of its two children and swap while that child wins. Stop when both children lose, or when there are none left.

Neither ever moves sideways, so both are bounded by the height of the tree. A complete tree of n nodes has height floor(log2 n): level d starts at index 2 ** d - 1, so the last index, n - 1, sits on level floor(log2 n). A million-item heap has height 19, so it holds 20 levels; a billion-item heap has height 29.

Watching it work

Start from the heap above, [2, 5, 3, 9, 7, 4]. Check rule 2 first: 2 beats 5 and 3, 5 beats 9 and 7, 3 beats 4. Valid.

Push 1. Append it, giving [2, 5, 3, 9, 7, 4, 1] with the new value at index 6.

• The parent of index 6 is (6 - 1) // 2 = 2, holding 3. 1 is smaller, so swap: [2, 5, 1, 9, 7, 4, 3], and the new value is at index 2.
• The parent of index 2 is (2 - 1) // 2 = 0, holding 2. 1 is smaller, so swap: [1, 5, 2, 9, 7, 4, 3].
• Index 0 is the root. Stop.

Two swaps in a seven-node heap, and nothing else moved.

Pop the minimum. The answer is 1, at index 0. Removing it leaves a hole at the root, and the fix is the part people get wrong: do not promote the smaller child, because that leaves a hole in the middle of the tree and breaks the shape rule. Take the last element instead — the only one whose removal keeps the shape legal — put it in the root, and sift it down.

• Remove the last element, 3, and drop it into the root: [3, 5, 2, 9, 7, 4].
• The children of index 0 are 5 and 2. The smaller, 2 at index 2, beats 3, so swap: [2, 5, 3, 9, 7, 4].
• The only child of index 2 is index 5, holding 4. That loses to 3, so nothing moves. Stop.

One swap, and the heap is back to the list it started as. On distinct values that always happens: the sift down retraces, in reverse, the path the sift up took.

The code

First, the versions without a heap

What you would otherwise reach for, both perfectly reasonable:

import bisect


def pop_smallest_by_scan(values: list[int]) -> int:
    """Remove and return the smallest item of an unordered list. O(n) a call."""
    position = min(range(len(values)), key=values.__getitem__)
    return values.pop(position)


def push_keeping_sorted(values: list[int], value: int) -> None:
    """Insert into a sorted list, so the smallest item is always values[0]."""
    bisect.insort(values, value)


unordered = [9, 4, 7, 1, 8]
print(pop_smallest_by_scan(unordered), unordered)

ordered: list[int] = []
for item in [9, 4, 7, 1, 8]:
    push_keeping_sorted(ordered, item)
print(ordered.pop(0), ordered)

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

Both are correct. Neither scales:

The sorted list looks better than it is: bisect.insort finds the insertion point in O(log n) but then shifts every later element one slot right, and that memory move is O(n). The heap is the one option that refuses to be O(n) at anything.

The heap itself

from collections.abc import Iterable


def sift_up(values: list[int], index: int) -> None:
    """Move values[index] up until its parent is no larger than it."""
    while index > 0:
        parent = (index - 1) // 2
        if values[parent] <= values[index]:
            # Every ancestor above the parent is already no larger than it,
            # so the whole path back to the root is now ordered.
            return
        values[parent], values[index] = values[index], values[parent]
        index = parent


def sift_down(values: list[int], index: int) -> None:
    """Move values[index] down until both of its children are no smaller."""
    size = len(values)
    while True:
        left = 2 * index + 1
        right = left + 1
        smallest = index

        if left < size and values[left] < values[smallest]:
            smallest = left
        if right < size and values[right] < values[smallest]:
            smallest = right
        if smallest == index:
            return

        values[index], values[smallest] = values[smallest], values[index]
        index = smallest


class MinHeap:
    """A binary min-heap kept in a flat list.

    The tree is implied by position, never stored: the children of index i
    live at 2 * i + 1 and 2 * i + 2, and the parent of index i lives at
    (i - 1) // 2. Every parent is no larger than either child, so the
    smallest item of the whole heap is always at index 0.
    """

    def __init__(self, items: Iterable[int] = ()) -> None:
        self.values: list[int] = list(items)
        # Floyd's bottom-up build: everything from len // 2 on is a leaf, and
        # a leaf is already a valid one-node heap.
        for index in range(len(self.values) // 2 - 1, -1, -1):
            sift_down(self.values, index)

    def __len__(self) -> int:
        return len(self.values)

    def peek(self) -> int:
        """The smallest item, without removing it. O(1)."""
        if not self.values:
            raise IndexError("peek from an empty heap")
        return self.values[0]

    def push(self, value: int) -> None:
        """Add a value and restore the heap property. O(log n)."""
        self.values.append(value)
        sift_up(self.values, len(self.values) - 1)

    def pop(self) -> int:
        """Remove and return the smallest item. O(log n)."""
        if not self.values:
            raise IndexError("pop from an empty heap")
        smallest = self.values[0]
        last = self.values.pop()
        if self.values:
            # One value now sits in the wrong place instead of a hole.
            self.values[0] = last
            sift_down(self.values, 0)
        return smallest


heap = MinHeap([9, 4, 7, 1, 8, 2, 6])
print("built:  ", heap.values)
print("peek:   ", heap.peek())
heap.push(3)
print("push 3: ", heap.values)
print("drained:", [heap.pop() for _ in range(len(heap))])

built:   [1, 4, 2, 9, 8, 7, 6]
peek:    1
push 3:  [1, 3, 2, 4, 8, 7, 6, 9]
drained: [1, 2, 3, 4, 6, 7, 8, 9]

Thirty lines of logic, no recursion, and one list.

How the code maps to the idea

pop moves the last element, not a child. self.values.pop() removes the final slot — the only removal the shape rule allows — and that value is written over the root. It is the most important line in the class: promoting a child instead would perforate the tree.

sift_up compares with <= and stops. Equal values are never swapped, so a push never pays for a move that changes nothing. Once the parent is no larger, every ancestor above it is no larger too, by induction along the path, so there is nothing left to check.

sift_down picks the smaller child first, then decides. Two comparisons per level, not one. Comparing the parent against only the left child is the classic broken sift down: it leaves a smaller right child sitting under a larger parent.

Both bounds checks matter. left < size fails at leaves; right < size fails at the one node with a left child and no right child, which exists whenever the heap holds an even number of items.

Both loops are iterative. A recursive sift down reads better and costs O(log n) stack frames; the loop costs nothing and cannot hit a recursion limit.

__init__ builds bottom-up, from the last non-leaf backwards to index 0. Backwards is not a style choice — sift_down assumes the subtrees under a node are already valid heaps, and decreasing index order is what guarantees that.

Edge cases fall out of the arithmetic. For an empty or one-item list len // 2 - 1 is at most −1, so range(-1, -1, -1) is empty and the build loop never runs; pop on a one-item heap removes the last element, finds self.values empty and skips the sift.

The same push and pop as the walkthrough, printing the list after every swap:

def show(label: str, values: list[int]) -> None:
    """One aligned line of the trace."""
    print(f"{label:<26}{values}")


def push_traced(values: list[int], value: int) -> None:
    """Push, printing the list after the append and after every swap."""
    values.append(value)
    index = len(values) - 1
    show(f"append {value} at index {index}", values)
    while index > 0:
        parent = (index - 1) // 2
        if values[parent] <= values[index]:
            return
        values[parent], values[index] = values[index], values[parent]
        index = parent
        show(f"sift {value} up to index {index}", values)


def pop_traced(values: list[int]) -> int:
    """Pop, printing the list after the root is refilled and after every swap."""
    smallest = values[0]
    last = values.pop()
    if not values:
        return smallest

    values[0] = last
    show(f"move {last} into the root", values)
    index = 0
    while True:
        left, right, target = 2 * index + 1, 2 * index + 2, index
        if left < len(values) and values[left] < values[target]:
            target = left
        if right < len(values) and values[right] < values[target]:
            target = right
        if target == index:
            show(f"popped {smallest}, heap valid", values)
            return smallest
        values[index], values[target] = values[target], values[index]
        index = target
        show(f"sift {last} down to index {index}", values)


live = [2, 5, 3, 9, 7, 4]
show("start", live)
push_traced(live, 1)
pop_traced(live)

start                     [2, 5, 3, 9, 7, 4]
append 1 at index 6       [2, 5, 3, 9, 7, 4, 1]
sift 1 up to index 2      [2, 5, 1, 9, 7, 4, 3]
sift 1 up to index 0      [1, 5, 2, 9, 7, 4, 3]
move 3 into the root      [3, 5, 2, 9, 7, 4]
sift 3 down to index 2    [2, 5, 3, 9, 7, 4]
popped 1, heap valid      [2, 5, 3, 9, 7, 4]

Every line matches the hand trace, including the return to the starting list.

Why building a heap is O(n)

There are two ways to turn n loose items into a heap, and they do not cost the same.

Build by pushing. Start empty and push n times. Each push costs at most the depth of the tree, so the total is at most n log₂ n.

Build bottom-up, the way __init__ does it — Robert Floyd's method from 1964. Sift down every non-leaf index, from n // 2 - 1 back to 0. The obvious bound is the same: about n/2 non-leaf nodes, each sift down costing up to log₂ n, therefore O(n log n). That bound is true but badly loose, and the real answer is O(n). It overshoots because it charges every node the price of the root — and there is only one root.

Count by height, not by node

A node's height is the number of edges on the longest path from it down to a leaf, so leaves have height 0 and the root of a complete n-node tree has height floor(log2 n). Two facts:

• A sift down starting at a node of height h costs at most h swaps, because each swap descends exactly one level.
• At most n / 2 ** (h + 1) nodes have height h, rounded up to a whole node. Half the nodes are leaves at height 0, a quarter are at height 1, an eighth at height 2 — each height up halves the population.

Multiply and sum over every height. The build costs at most the sum, over h from 0 to log₂ n, of (n / 2 ** (h + 1)) * h. Pull the n/2 out front and what is left is the sum of h / 2 ** h, a convergent series with a tidy value, provable with nothing but subtraction:

S      =       1/2 + 2/4 + 3/8 + 4/16 + ...
2S     = 1   + 2/2 + 3/4 + 4/8 + 5/16 + ...
2S - S = 1   + 1/2 + 1/4 + 1/8 + 1/16 + ...  = 2

Doubling S shifts every term one place left; subtracting the original leaves a plain geometric series that sums to 2. So S = 2, and the build costs at most (n / 2) * 2 = n swaps. Linear, with a constant of 1. The rounding dropped on the way costs nothing: the exact worst case is the sum of every node's height, and on a complete tree that always lands a little under n.

In English, without the algebra: the nodes that can fall a long way are rare, and the nodes that are plentiful can barely fall at all. Half the array is leaves, and a leaf cannot move. Only one node can fall the full 19 levels of a million-item heap.

The push build is the mirror image

The same counting, run the other way, explains why building by pushing is genuinely slower. A sift up is bounded by a node's depth, and depth is the opposite of height: the bottom level holds half the nodes at height 0 and maximum depth. The push build charges its highest price to the most populous level, which is exactly the mistake the bottom-up build avoids.

Summing depth * 2 ** depth over all 20 levels of a perfect 1,048,575-node heap gives 18,874,370 swaps against the bottom-up bound of 1,048,555 — same heap, eighteen times the work. That worst case is reachable: push values in strictly decreasing order and every new item is the new minimum, so every push travels all the way to the root. Both builds, measured on a perfect tree of 65,535 items:

def build_by_pushes(items: list[int]) -> int:
    """Build a heap by pushing one item at a time; return the swaps used."""
    values: list[int] = []
    swaps = 0
    for item in items:
        values.append(item)
        index = len(values) - 1
        while index > 0:
            parent = (index - 1) // 2
            if values[parent] <= values[index]:
                break
            values[parent], values[index] = values[index], values[parent]
            index = parent
            swaps += 1
    return swaps


def build_bottom_up(items: list[int]) -> int:
    """Floyd's build: sift every non-leaf down; return the swaps used."""
    values = list(items)
    size = len(values)
    swaps = 0
    for start in range(size // 2 - 1, -1, -1):
        index = start
        while True:
            left, right, target = 2 * index + 1, 2 * index + 2, index
            if left < size and values[left] < values[target]:
                target = left
            if right < size and values[right] < values[target]:
                target = right
            if target == index:
                break
            values[index], values[target] = values[target], values[index]
            index = target
            swaps += 1
    return swaps


def shuffled(size: int) -> list[int]:
    """A deterministic shuffle, so the counts below reproduce anywhere."""
    values = list(range(size))
    state = 20240607
    for index in range(size - 1, 0, -1):
        # The low bits of a plain LCG repeat far too fast to shuffle with, so
        # mix the whole word before taking a remainder.
        state = (6364136223846793005 * state + 1442695040888963407) % (1 << 64)
        other = (state ^ (state >> 33)) % (index + 1)
        values[index], values[other] = values[other], values[index]
    return values


n = 2 ** 16 - 1
cases = [
    ("descending", list(range(n, 0, -1))),
    ("shuffled", shuffled(n)),
    ("ascending", list(range(n))),
]

print(f"n = {n:,} items, a perfect tree of 16 levels")
print(f"{'input order':<12}{'push build':>28}{'bottom-up build':>28}")
for label, data in cases:
    up, down = build_by_pushes(data), build_bottom_up(data)
    print(f"{label:<12}{up:>12,} swaps ({up / n:5.2f} n)"
          f"{down:>12,} swaps ({down / n:5.2f} n)")

levels = 20
total_nodes = 2 ** levels - 1
push_worst = sum(depth * 2 ** depth for depth in range(levels))
bottom_up_bound = sum((levels - 1 - depth) * 2 ** depth for depth in range(levels))
print(f"\nprojected to n = {total_nodes:,} ({levels} levels)")
print(f"  push build, worst case: {push_worst:>12,} swaps = {push_worst / total_nodes:5.2f} n")
print(f"  bottom-up build, bound: {bottom_up_bound:>12,} swaps = {bottom_up_bound / total_nodes:5.2f} n")

n = 65,535 items, a perfect tree of 16 levels
input order                   push build             bottom-up build
descending       917,506 swaps (14.00 n)      65,519 swaps ( 1.00 n)
shuffled          84,141 swaps ( 1.28 n)      48,760 swaps ( 0.74 n)
ascending              0 swaps ( 0.00 n)           0 swaps ( 0.00 n)

projected to n = 1,048,575 (20 levels)
  push build, worst case:   18,874,370 swaps = 18.00 n
  bottom-up build, bound:    1,048,555 swaps =  1.00 n

The descending row hits the predicted worst case exactly: 917,506 is sum(depth * 2 ** depth) over 16 levels, and 65,519 is exactly n - 16, the sum of the heights of all 65,535 nodes and so the most any sift-down build can do. Neither bound is approximate. The ascending row is free for both, since ascending input is already a valid min-heap.

The shuffled row is the honest caveat. On randomly ordered input the push build costs only 1.28 swaps per item, because a random new value usually loses to its parent after a swap or two — that O(n log n) is a worst case, not a typical one. What the bottom-up build adds is a guarantee of under n swaps on every input, which is why heapq.heapify exists and why you should use it instead of a loop of pushes whenever you already hold the data.

Complexity

Two rows deserve more than a line.

push is O(log n) in the worst case but O(1) on average. Travelling far needs the new value to beat every ancestor; on random input two pushes in three stop after at most one swap. The measured 1.28 swaps per item above is that average, and it does not grow with n.

pop has no such luck; it really does pay close to log n every time. The value promoted into the root was, one instant earlier, the last leaf — and in a min-heap a leaf is at least as large as all of its roughly log₂ n ancestors. A large value at the top sinks nearly all the way back down, and no input makes that cheap.

So pushing n items and immediately draining them costs O(n log n). That is a sorting algorithm — heap sort — and not a faster one than sorted(). A heap earns its place when pushes and pops are interleaved with other work.

The real tool: heapq

Nothing above needs to be written by hand. Python ships heapq, a min-heap over an ordinary list, implemented in C with a pure-Python fallback. There is no heap class: the functions operate directly on a list you own.

import heapq

queue = [9, 4, 7, 1, 8, 2, 6]
heapq.heapify(queue)
print("heapify:    ", queue)

heapq.heappush(queue, 3)
print("push 3:     ", queue)
print("pop:        ", heapq.heappop(queue), queue)
print("replace 5:  ", heapq.heapreplace(queue, 5), queue)
print("pushpop 0:  ", heapq.heappushpop(queue, 0), queue)
print("3 smallest: ", heapq.nsmallest(3, queue))
print("3 largest:  ", heapq.nlargest(3, queue))

heapify:     [1, 4, 2, 9, 8, 7, 6]
push 3:      [1, 3, 2, 4, 8, 7, 6, 9]
pop:         1 [2, 3, 6, 4, 8, 7, 9]
replace 5:   2 [3, 4, 6, 5, 8, 7, 9]
pushpop 0:   0 [3, 4, 6, 5, 8, 7, 9]
3 smallest:  [3, 4, 5]
3 largest:   [9, 8, 7]

heapify produced the same list as the hand-written MinHeap, which is no coincidence: it is Floyd's build. On distinct values CPython's heappop also leaves exactly the array the textbook sift down leaves. With duplicates the two can differ, because CPython's internal _siftup follows the smaller child all the way to the bottom and then sifts the displaced value back up — fewer comparisons, and a different but equally valid arrangement of the ties.

The two fused operations each save a full pass:

• heapreplace(heap, x) pops the smallest and then pushes x, in one sift down. The returned value is never x; above it returned 2 and left 5 behind.
• heappushpop(heap, x) pushes x and then pops, so if x is below the current root it comes straight back out and the heap is untouched — exactly what happened with 0.

heapq only builds min-heaps

There is no max flag. Two standard workarounds:

Negate the values. Push -value and negate again on the way out. Numbers only — there is nothing to negate if the priority is a string or a date.

Push tuples. Python compares tuples element by element, so (priority, payload) orders by priority first; for a max-heap push (-priority, payload).

The tuple has a trap that catches everyone exactly once. When two priorities tie, the comparison falls through to the payload — and if the payload has no ordering, you get a TypeError from inside an unrelated push:

import itertools

scores = [("ana", 71), ("ben", 94), ("cal", 58)]
max_heap = [(-score, name) for name, score in scores]
heapq.heapify(max_heap)
best_score, best_name = heapq.heappop(max_heap)
print(f"highest score: {best_name} with {-best_score}")


class Task:
    """A payload with no ordering of its own -- the usual real-world case."""

    def __init__(self, name: str) -> None:
        self.name = name


broken: list[tuple[int, Task]] = []
try:
    heapq.heappush(broken, (1, Task("deploy")))
    heapq.heappush(broken, (1, Task("backup")))
except TypeError as error:
    print(f"{type(error).__name__}: {error}")

sequence = itertools.count()
fixed: list[tuple[int, int, Task]] = []
for priority, name in [(2, "email"), (1, "deploy"), (1, "backup")]:
    heapq.heappush(fixed, (priority, next(sequence), Task(name)))
print("order out:", [heapq.heappop(fixed)[2].name for _ in range(3)])

highest score: ben with 94
TypeError: '<' not supported between instances of 'Task' and 'Task'
order out: ['deploy', 'backup', 'email']

The fix is a monotonically increasing counter as the second tuple element, from itertools.count(). No two entries can tie on it, so the payload is never compared, and as a bonus equal priorities come out first-in-first-out: deploy was pushed before backup and left first. The alternative is a dataclass with order=True and field(compare=False) on every field that should not take part.

What heapq will not do at all

There is no decrease_key and no remove. Changing an item's priority after it is in the heap re-sorts nothing, and finding it to remove it is an O(n) scan. The workaround in the heapq documentation's own implementation notes is lazy deletion: keep a dictionary from task to heap entry, blank an outdated entry's payload with a sentinel, push the replacement, and skip sentinels as you pop. Python implementations of Dijkstra do a simpler version — push a new (distance, node) pair and ignore any popped pair whose distance is already stale.

Two more limits. heapq works on a bare list, so nothing stops anyone calling queue.append(...) and quietly corrupting the invariant. And it is not thread-safe: across threads use queue.PriorityQueue, which wraps heapq in a lock, or asyncio.PriorityQueue inside an event loop.

Selecting the top k

Finding the 10 largest of 100,000 items does not need a sort. Keep a min-heap of size 10: the root is the weakest of your current winners, so a new item only matters if it beats the root.

def top_k(items: list[int], k: int) -> list[int]:
    """The k largest items, descending, using a min-heap of at most k entries."""
    keepers: list[int] = []
    for item in items:
        if len(keepers) < k:
            heapq.heappush(keepers, item)
        elif item > keepers[0]:
            # keepers[0] is the weakest survivor, so this one replaces it.
            heapq.heapreplace(keepers, item)
    return sorted(keepers, reverse=True)


stream = shuffled(100_000)
print("top 5:", top_k(stream, 5))
print("nlargest agrees:", top_k(stream, 5) == heapq.nlargest(5, stream))

keepers: list[int] = []
heap_operations = 0
for value in stream:
    if len(keepers) < 10:
        heapq.heappush(keepers, value)
        heap_operations += 1
    elif value > keepers[0]:
        heapq.heapreplace(keepers, value)
        heap_operations += 1
print(f"heap touched {heap_operations} times out of {len(stream):,} items")

top 5: [99999, 99998, 99997, 99996, 99995]
nlargest agrees: True
heap touched 123 times out of 100,000 items

The cost is O(n log k): every item costs one comparison against keepers[0], and only survivors pay a O(log k) sift. Sorting the whole list is O(n log n) — for n = 100,000 and k = 10 that is log₂ k of 3.3 against log₂ n of 17.

The measured line makes the point better than the big-O does. Only 123 items out of 100,000 caused any heap work; the other 99,877 lost one comparison and were dropped. That count grows like k * ln(n / k), so it stays tiny on far larger streams — and since the heap never holds more than k items, this works on input that does not fit in memory.

heapq.nlargest does exactly this, tie-broken with a decreasing counter, and it is what collections.Counter.most_common(k) calls. Two shortcuts in its source are worth copying: when k is 1 it uses max(), and when k is at least the input length it just sorts. A heap only wins while k stays small next to n.

Merging k sorted lists

You have k sorted sequences totalling N items and want one sorted output. Concatenating and sorting costs O(N log N) and needs everything in memory at once. A heap holds only the current head of each list — k items, never more.

def merge_sorted(lists: list[list[int]]) -> list[int]:
    """Merge k sorted lists into one sorted list, with a heap of k entries."""
    frontier = [(values[0], which, 0) for which, values in enumerate(lists) if values]
    heapq.heapify(frontier)

    merged: list[int] = []
    while frontier:
        value, which, position = heapq.heappop(frontier)
        merged.append(value)
        position += 1
        if position < len(lists[which]):
            heapq.heappush(frontier, (lists[which][position], which, position))
    return merged


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

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

Each of the N items is pushed once and popped once, and each of those operations acts on a heap of at most k entries, so the total is O(N log k). The list index in the tuple does double duty: it tells the loop which list to refill from, and it breaks ties between equal values.

heapq.merge does this for you and returns a lazy iterator, so the inputs can be files or generators far larger than memory. That is the shape of the final phase of an external merge sort: sort as many chunks as fit in RAM, write each to disk, then merge the runs with a k-entry heap.

When to use it, and when not to

Use a heap when you repeatedly need the extreme item of a collection that keeps changing. The signature is a loop that pulls the best-so-far, does work, and pushes new candidates: schedulers ordering by time, Dijkstra and A* ordering a frontier by cost, Huffman coding pulling the two rarest symbols, top-k over a stream.

Do not use one to sort. Push-then-drain is O(n log n) with far worse constants than sorted(), which is C-level Timsort. And when k approaches n, sorting once beats n pops anyway.

Do not use one to look things up. A heap has no ordering between subtrees, so value in heap is a linear scan. Use a dict or a set.

Do not use one when you need order, not just the minimum. heap[1] is not the second smallest — that is whichever of heap[1] and heap[2] is smaller. There is no in-order traversal, no successor, no range query. For those, use a sorted list with bisect or a balanced binary search tree.

Be wary when priorities change often. Lazy deletion works, and is what most Python code does, but stale entries stay in the heap: memory grows with the number of updates rather than the number of live items. An indexed heap that tracks each item's position is the fix.

The binary heap's own weakness is memory locality. The children of index i sit at 2 * i + 1, so the parent-to-child jump grows with the index; near the bottom of a large heap one step down the tree is a jump of megabytes, and nearly every step is a cache miss. A 4-ary heap, children at 4 * i + 1 through 4 * i + 4, halves the depth and often measures faster despite more comparisons per level. Fibonacci heaps improve the theory — O(1) amortised decrease-key — and lose badly in practice on constant factors.

Where it shows up in the real world

Python's own standard library. The sched module keeps its event queue as a heap and pops the earliest deadline. asyncio's event loop keeps every scheduled timer in self._scheduled, a heap of timer handles pushed with heapq.heappush and popped with heapq.heappop. collections.Counter.most_common(k) is a direct call to heapq.nlargest.

Dijkstra's algorithm and A*. The frontier is a priority queue keyed on distance-so-far. Scanning an array for the nearest unvisited node makes Dijkstra O(V²); a binary heap makes it O((V + E) log V), the difference between usable and not on a road network.

Huffman coding. Put every symbol in a min-heap keyed on frequency, then repeat: pop the two smallest, push a node holding their combined frequency. For n symbols that is exactly 2 * (n - 1) pops and n - 1 pushes, and the tree that falls out is the optimal prefix code.

Discrete event simulation. A simulation of a queue, a network or a factory floor keeps a future event list ordered by event time: the loop pops the next event, jumps the clock to it, and that event schedules further events at later times. SimPy, the most widely used Python simulation package, keeps its event queue in a heapq. The pattern fits in fifteen lines:

event_queue: list[tuple[int, int, str]] = []
tie_breaker = itertools.count()


def schedule(time: int, name: str) -> None:
    """Put a future event on the queue; it comes back out in time order."""
    heapq.heappush(event_queue, (time, next(tie_breaker), name))


schedule(5, "sensor A reading")
schedule(2, "sensor B reading")
schedule(9, "flush log")

while event_queue:
    now, _, event = heapq.heappop(event_queue)
    print(f"t={now:>2}  {event}")
    if event == "sensor B reading" and now < 8:
        schedule(now + 4, "sensor B reading")

t= 2  sensor B reading
t= 5  sensor A reading
t= 6  sensor B reading
t= 9  flush log
t=10  sensor B reading

Events were created out of order and delivered in time order, and the repeating sensor scheduled its own next reading mid-run. That is the whole architecture of a simulation engine.

Other languages. Java's java.util.PriorityQueue is a binary heap over an array; C++'s std::priority_queue is a binary heap over a vector, with the raw operations exposed as std::push_heap and std::pop_heap.

Common mistakes

Promoting a child instead of the last leaf on pop. It leaves a gap in the middle of the tree, so the shape rule breaks and the index arithmetic stops describing the structure. Move the last element into the root and sift it down.

Comparing the parent to only the left child in a hand-rolled sift down. It looks fine on small examples and returns wrong minima later.

Building with a loop of heappush when you already hold the data. Use heapq.heapify: guaranteed O(n) instead of O(n log n) worst case, and one line instead of a loop.

Pushing (priority, object) without a tie-break. It works until two priorities are equal, then raises a TypeError from inside heappush — usually in production. Push (priority, next(counter), object).

Assuming heap[1] is the second smallest. Only index 0 is guaranteed; the runner-up is the smaller of indices 1 and 2.

Mutating an item already in the heap. Changing the field the ordering depends on does not move the item, and the invariant is silently broken from then on. Push a new entry and invalidate the old one.

Calling heappop on an empty list. It raises IndexError rather than returning None.

Practice

1. Write is_min_heap(values) returning whether a list obeys the order rule, in a single O(n) pass over the parents only.
2. Add pop_second_smallest to MinHeap — it must find the answer by looking at indices 1 and 2, not by scanning.
3. Turn MinHeap into a max-heap by changing only the comparison operators, then implement the same thing again by negating on the way in and out. Compare which one you would rather maintain.
4. Keep a running median of a stream using two heaps: a max-heap of the lower half and a min-heap of the upper half, rebalanced so their sizes never differ by more than one.
5. Implement lazy deletion: a priority queue with push, pop and remove(task) that marks removed entries with a sentinel and skips them on the way out. Count how many stale entries build up over 1,000 priority updates.

Summary

A heap is the smallest structure that keeps a minimum available while the data keeps changing. Two rules — complete shape, parent no larger than its children — collapse a binary tree into a flat array where the children of index i sit at 2i + 1 and 2i + 2, and every operation walks one path of length log₂ n. The build is the part worth remembering: bottom-up it costs under n swaps, because the nodes that can fall far are rare and the nodes that are plentiful cannot move at all. In Python you write heapq, remember it only does minimums, and always push a tie-breaking counter alongside the payload.

Keep reading

• Heap Sort in Python — the same structure turned into an in-place O(n log n) sort with O(1) extra memory.
• Dijkstra's Algorithm — the priority queue doing the work that makes shortest paths fast.
• Huffman Coding — building an optimal compression code by popping the two smallest frequencies.
• Binary Search Trees — the structure to reach for when you need order, not just the minimum.
• Big O Notation — the summation machinery behind the O(n) build proof.