Heaps and priority queues

Theo's packing queue from lesson 3 serves orders strictly in arrival order. That was fine until Ridgeline started selling express shipping. Now an express order that arrives at 10:40 must be packed before a standard order that arrived at 9:15, and wholesale orders due today come before both. The next order to pack is no longer the oldest one but the one with the highest priority. A structure that always hands back the most urgent item, while new items keep arriving, is called a priority queue.

This lesson shows how to build a priority queue efficiently with a binary heap, a clever tree stored inside a plain list. You will learn how a heap keeps its smallest item at the front with O(log n) insertions and removals, use Python's heapq module, handle ties and maximum-first ordering, and apply heaps to some of the most common practical problems: processing work by priority, finding the top k items in a large dataset and merging sorted streams.

Why not a sorted list?

A priority queue needs two operations: insert an item with a priority, and remove the item with the best priority. One obvious approach keeps a list sorted by priority. Removing the best item is then easy, but every insertion must find its place and shift the items after it, O(n). Another approach appends new items in O(1) but scans the whole list to find the best one, also O(n). A heap gets both operations down to O(log n), which for thousands of waiting orders is the difference between a few steps and thousands.

The binary heap

A binary heap is a binary tree with two rules. The shape rule says the tree is complete: every level is full except possibly the last, which fills from left to right. The heap rule says every node is less than or equal to its children. Together they guarantee that the smallest item is always at the root, while saying nothing about the order of brothers and sisters, which is what makes heaps cheap to maintain: a heap is only partly sorted.

Because the tree is complete, it can be stored in a list with no gaps and no references. The root is at index 0, and the children of the node at index i are at 2i + 1 and 2i + 2, so the parent of the node at index i is at (i - 1) // 2. Moving around the tree is just arithmetic on indexes.

Inserting an item appends it at the end of the list, the next free spot in the bottom level, then lets it sift up: while it is smaller than its parent, swap the two. Removing the smallest item takes the root, moves the last item into the root's place, then lets it sift down: while it is larger than its smaller child, swap them. Each sift moves one level per step, and a complete tree of n items has about log₂ n levels, so both operations are O(log n).

class MinHeap:
    def __init__(self):
        self.a = []

    def push(self, item):
        a = self.a
        a.append(item)
        i = len(a) - 1
        while i > 0 and a[i] < a[(i - 1) // 2]:            # sift up
            p = (i - 1) // 2
            a[i], a[p] = a[p], a[i]
            i = p

    def pop(self):
        a = self.a
        smallest, last = a[0], a.pop()
        if a:
            a[0], i = last, 0
            while True:                                    # sift down
                c = 2 * i + 1
                if c >= len(a):
                    break
                if c + 1 < len(a) and a[c + 1] < a[c]:
                    c += 1                                 # the smaller child
                if a[i] <= a[c]:
                    break
                a[i], a[c] = a[c], a[i]
                i = c
        return smallest

h = MinHeap()
for p in [7, 3, 9, 1, 4, 5, 2, 8]:
    h.push(p)
print("stored as:", h.a)
print("popped:", [h.pop() for _ in range(8)])

stored as: [1, 3, 2, 7, 4, 9, 5, 8]
popped: [1, 2, 3, 4, 5, 7, 8, 9]

The heapq module

Python's heapq module provides these operations as functions that work on an ordinary list, written in C for speed. heappush adds an item, heappop removes and returns the smallest, and heap[0] peeks at the smallest without removing it. heapify turns an existing list into a heap in place in O(n), faster than pushing the items one by one. heapq always builds a min-heap: the smallest item comes out first.

import heapq

stock = [42, 15, 27, 9, 12, 30, 21]
heapq.heapify(stock)                     # O(n), in place
print("smallest:", stock[0])
heapq.heappush(stock, 5)
print([heapq.heappop(stock) for _ in range(3)])
print("left:", sorted(stock))

smallest: 9
[5, 9, 12]
left: [15, 21, 27, 30, 42]

Priorities, ties and maximum first

Real priority queues store items with priorities, usually as tuples, because tuples compare element by element. Theo uses (priority, arrival, order_id), where priority 0 is wholesale due today, 1 is express and 2 is standard. The arrival number breaks ties so that equal priorities come out first in, first out, and it also prevents Python from ever comparing the order objects themselves, which might not support comparison. For a maximum-first queue, push the negated priority, since the smallest negative number corresponds to the largest value.

import heapq, itertools

arrival = itertools.count()              # 0, 1, 2, ... a tie-breaker in arrival order
queue = []

def add(order_id, priority):
    heapq.heappush(queue, (priority, next(arrival), order_id))

add(1041, 2); add(1042, 2); add(1043, 1); add(1044, 0); add(1045, 1)
while queue:
    priority, _, order_id = heapq.heappop(queue)
    print(f"pack {order_id} (priority {priority})")

biggest_first = []
for total in [41.5, 18.5, 64.25, 22.0]:
    heapq.heappush(biggest_first, -total)
print([-heapq.heappop(biggest_first) for _ in range(2)])

pack 1044 (priority 0)
pack 1043 (priority 1)
pack 1045 (priority 1)
pack 1041 (priority 2)
pack 1042 (priority 2)
[64.25, 41.5]

def packing_order(orders):
    return [order_id for order_id, priority in orders]

print(packing_order([(1041, 2), (1042, 1), (1043, 2), (1044, 0)]))

Top k with a heap

Rosa wants the ten biggest loyalty spenders out of 50,000 members for a thank-you mailing. Sorting all 50,000 is O(n log n) and does more work than needed. heapq.nlargest(k, items, key=...) keeps a min-heap of just the best k seen so far: each new member is compared with the smallest of the current top k at the root, and only replaces it if bigger. That costs O(n log k), and with k = 10, log k is tiny, so it is nearly a single pass. heapq.nsmallest works the same way for the smallest items.

import heapq, random

random.seed(201)
members = [(f"M{k:05d}", round(random.uniform(5, 900), 2)) for k in range(50_000)]
top = heapq.nlargest(5, members, key=lambda m: m[1])
for member, spent in top:
    print(member, spent)

def top_k(items, k):
    """The k largest values, using a min-heap of size k."""
    heap = []
    for x in items:
        if len(heap) < k:
            heapq.heappush(heap, x)
        elif x > heap[0]:
            heapq.heapreplace(heap, x)     # pop the smallest and push x in one step
    return sorted(heap, reverse=True)

print(top_k([spent for _, spent in members], 3) == [s for _, s in top[:3]])

M35591 899.95
M16633 899.93
M12632 899.85
M30076 899.84
M00652 899.8
True

def lowest_stock(stock, k):
    return list(stock)[:k]

print(lowest_stock({'RC-101': 42, 'RC-104': 9, 'RC-105': 12, 'RC-102': 15}, 2))

Changing priorities

Real queues change while items wait. A customer upgrades an order to express shipping, or cancels it altogether. A heap has no fast way to find an item in the middle of its list, since only the root is in a known place, so searching for the order and repairing the heap would cost O(n). The standard technique is lazy deletion: instead of removing the old entry, mark it as canceled and push a new entry if the priority changed. When a canceled entry reaches the top, it is simply thrown away. A dictionary from order id to its current entry makes the marking O(1).

import heapq, itertools

class OrderQueue:
    CANCELLED = None

    def __init__(self):
        self.heap, self.entries, self.count = [], {}, itertools.count()

    def add(self, order_id, priority):
        if order_id in self.entries:
            self.cancel(order_id)                   # changing priority: retire the old entry
        entry = [priority, next(self.count), order_id]
        self.entries[order_id] = entry
        heapq.heappush(self.heap, entry)

    def cancel(self, order_id):
        entry = self.entries.pop(order_id)
        entry[2] = self.CANCELLED                   # mark it; it stays in the heap for now

    def next(self):
        while self.heap:
            priority, _, order_id = heapq.heappop(self.heap)
            if order_id is not self.CANCELLED:
                del self.entries[order_id]
                return order_id
        return None

q = OrderQueue()
for order_id in (1041, 1042, 1043):
    q.add(order_id, 2)                              # all standard shipping
q.add(1043, 1)                                      # 1043 upgrades to express
q.cancel(1041)                                      # 1041 is cancelled
print(q.next(), q.next(), q.next())

1043 1042 None

Each entry is a list rather than a tuple so it can be marked in place, and the heap still compares entries by priority and then arrival number, never reaching the order id. The cost is some wasted space for canceled entries until they surface, which is usually a fair trade. The Python documentation for heapq describes this same pattern, and lesson 11 relies on a similar idea when it finds shortest routes.

Merging sorted streams

Heaps also merge several sorted sequences efficiently. Each of Ridgeline's three sales channels, the cafe till, the online shop and wholesale, writes a log already sorted by time, and Anita wants one combined log. Put the first entry of each log in a heap; repeatedly pop the earliest entry and push the next entry from the same log. With k logs and n entries in total, that is O(n log k). heapq.merge does exactly this lazily, returning an iterator, so even logs too big for memory can be merged as they are read.

import heapq

cafe = [("08:05", "cafe", 5.25), ("09:40", "cafe", 3.25), ("11:10", "cafe", 9.00)]
online = [("08:30", "online", 41.50), ("10:15", "online", 18.50)]
wholesale = [("09:00", "wholesale", 640.00)]

for time, channel, amount in heapq.merge(cafe, online, wholesale):
    print(time, f"{channel:<9}", f"{amount:>7.2f}")

08:05 cafe         5.25
08:30 online      41.50
09:00 wholesale  640.00
09:40 cafe         3.25
10:15 online      18.50
11:10 cafe         9.00

Two related ideas complete the picture. Heap sort builds a heap and pops everything, an O(n log n) sort that needs no extra memory, though Python's sorted is faster in practice. And heaps sit inside many famous algorithms, including the shortest-route algorithm previewed in lesson 11, schedulers that decide which task an operating system runs next, and event-driven simulations that always process the earliest upcoming event, a natural upgrade of the minute-by-minute cafe simulation in lesson 3.

def merge_logs(logs):
    out = []
    for log in logs:
        out.extend(log)
    return out

print(merge_logs([[('08:05', 5.25), ('09:40', 3.25)], [('08:30', 41.5)]]))