Priority Queues in Practice

A priority queue is a data structure that always hands you the highest-priority item next, no matter what order items were added in — think of a hospital emergency room, where a patient having a heart attack is treated before someone who arrived earlier with a sprained ankle. In Python, priority queues are almost always built on a binary heap, and the standard library’s heapq module gives you exactly that: an efficient, array-based min-heap. This lesson covers not just the theory but how heapq is actually used in real code — task schedulers, Dijkstra’s shortest-path algorithm, and the tie-breaking tricks you need once two items share a priority.

Overview: How Priority Queues Work

A priority queue is an abstract data type (ADT): it supports insert and extract the most important item, and nothing says how that has to be implemented under the hood. In Python, the concrete implementation almost everyone reaches for is a binary heap via the heapq module. A binary heap stores its elements in a plain list but treats that list as a complete binary tree: for the element at index i, its children live at indices 2i + 1 and 2i + 2, and its parent lives at index (i - 1) // 2. The tree is ‘complete’ because every level fills up left to right before a new level starts, which is exactly why a flat list can represent it with no wasted space and no pointers.

The heap holds one invariant, called the min-heap property: every parent is less than or equal to both of its children. Notice what this does not guarantee — it says nothing about the relationship between siblings, or between any two nodes that aren’t in a direct parent-child line. That weaker guarantee is exactly why a heap is cheaper to maintain than a fully sorted list: restoring the property after an insert or removal only ever touches one path from root to leaf, which has length O(log n), instead of re-sorting everything.

Min-Heap vs Max-Heap

heapq only implements a min-heap — the smallest value always pops first. If you want the largest value first (a max-heap), there is no built-in flag for that. The standard trick is to negate every value on the way in and negate it again on the way out, since the smallest negative number corresponds to the largest original number. When priorities are more complex than plain numbers, store a tuple like (priority, item) and negate just the priority.

Alternative Implementations

A heap isn’t the only way to build a priority queue — it’s just the best general-purpose balance between insert cost and extract cost.

Implementation Insert Extract-min Notes
Unsorted list O(1) O(n) Insert is a plain append; finding the minimum means scanning every element
Sorted list O(n) O(1) The minimum is always at the front, but inserting means shifting elements to keep order
Binary heap (heapq) O(log n) O(log n) Balances both operations — the standard choice for a general-purpose priority queue

Time and Space Complexity

Every heap operation’s cost comes down to one fact: a binary heap holding n elements has height O(log n), because each level of a complete binary tree roughly doubles the number of nodes.

Operation Complexity Why
heapq.heappush O(log n) The new element is appended, then swapped upward (‘sifted up’) at most once per level of the tree
heapq.heappop O(log n) The root is removed, the last element takes its place, then it’s sifted down through at most O(log n) levels to restore the heap property
heap[0] (peek) O(1) The min-heap invariant guarantees the smallest element is always at index 0
heapq.heapify O(n) Building a heap from an existing list beats n individual pushes because most nodes sit near the bottom of the tree and need little or no sifting
heapq.nlargest(k, it) / nsmallest O(n log k) Scans all n items while maintaining a heap of only size k

Space complexity is O(n) — the heap is stored as a single flat Python list with no per-node pointer overhead, unlike a pointer-based tree structure.

Examples

Example 1: Basic heap operations

This example builds a heap from an unordered list with heapify, then pops from it twice, printing the underlying array after each step so you can see the heap reorganize itself.

import heapq

numbers = [5, 3, 8, 1, 9, 2]
heapq.heapify(numbers)
print('heapified:', numbers)

first = heapq.heappop(numbers)
print('popped:', first, '-> heap:', numbers)

second = heapq.heappop(numbers)
print('popped:', second, '-> heap:', numbers)

Output:

heapified: [1, 3, 2, 5, 9, 8]
popped: 1 -> heap: [2, 3, 8, 5, 9]
popped: 2 -> heap: [3, 5, 8, 9]

heapify rearranges [5, 3, 8, 1, 9, 2] into [1, 3, 2, 5, 9, 8]. Check the min-heap property by hand: index 0 (value 1) is less than or equal to its children at indices 1 and 2 (3 and 2); index 1 (value 3) is less than or equal to its children at indices 3 and 4 (5 and 9); index 2 (value 2) is less than or equal to its child at index 5 (8). Every parent-child pair holds, so it’s a valid heap — even though the list as a whole is clearly not sorted. Each heappop call removes the root (the current minimum), moves the last element into its place, and sifts it down until the property is restored again.

Example 2: A task scheduler with tie-breaking

A common real use of a priority queue is running tasks in priority order. Here each task is a (priority, name) tuple; a lower number means it runs sooner.

import heapq


def schedule_tasks(tasks: list[tuple[int, str]]) -> list[str]:
    heap: list[tuple[int, str]] = []
    for priority, name in tasks:
        heapq.heappush(heap, (priority, name))

    order: list[str] = []
    while heap:
        priority, name = heapq.heappop(heap)
        order.append(name)
    return order


tasks = [(3, 'backup'), (1, 'deploy'), (2, 'test'), (1, 'alert')]
print(schedule_tasks(tasks))

Output:

['alert', 'deploy', 'test', 'backup']

Two tasks share priority 1: 'deploy' and 'alert'. When tuples compare equal on their first element, Python falls back to comparing the second element — so heapq compares the strings 'deploy' and 'alert' alphabetically, and 'alert' sorts first. That’s why the final order is alert, deploy, test, backup rather than deploy coming before alert just because it was pushed first. Popping every element off a heap one at a time like this is effectively heapsort, which is why the result comes out fully sorted by priority.

Example 3: Dijkstra’s shortest-path algorithm

The single most common ‘real’ use of a priority queue in algorithms is Dijkstra’s algorithm, which always expands the closest not-yet-finalized node next — exactly what a min-heap gives you for free.

import heapq


def dijkstra(graph: dict[str, list[tuple[str, int]]], source: str) -> dict[str, float]:
    distances: dict[str, float] = {node: float('inf') for node in graph}
    distances = 0
    priority_queue: list[tuple[float, str]] = [(0, source)]

    while priority_queue:
        current_distance, current_node = heapq.heappop(priority_queue)

        if current_distance > distances[current_node]:
            continue

        for neighbor, weight in graph[current_node]:
            distance = current_distance + weight
            if distance < distances[neighbor]:
                distances[neighbor] = distance
                heapq.heappush(priority_queue, (distance, neighbor))

    return distances


graph = {
    'A': [('B', 4), ('C', 1)],
    'B': [('D', 1)],
    'C': [('B', 2), ('D', 5)],
    'D': [],
}

result = dijkstra(graph, 'A')
print(result)

Output:

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

Trace it: starting from A with distance 0, we push B (distance 4) and C (distance 1). The heap always hands us the smallest next, so we pop C (1) before B. From C we find a cheaper route to B (1 + 2 = 3, better than 4) and push it, plus a route to D (1 + 5 = 6). Next we pop the new, cheaper B entry (3), and from there find an even cheaper D (3 + 1 = 4). When the stale B entry (4) and stale D entry (6) eventually get popped, the if current_distance > distances[current_node]: continue line detects they’re outdated and skips them — this is called lazy deletion, and it’s the standard way to handle ‘decrease priority’ with heapq, since the module has no built-in way to update an item already inside the heap.

How It Works Step by Step

Walk through Example 1’s heapify and two pops using the array-as-tree mapping (child indices 2i + 1 and 2i + 2, parent index (i - 1) // 2):

  • Start: [5, 3, 8, 1, 9, 2]. As a tree, 5 is the root, with 3 and 8 as its children, and 1, 9, 2 as grandchildren.
  • After heapify: [1, 3, 2, 5, 9, 8]. The global minimum, 1, has bubbled up to the root; every parent is still less than or equal to its children.
  • First heappop: index 0 (value 1) is saved to return. The last element, 8, moves into index 0, giving [8, 3, 2, 5, 9]. 8 is compared against its children (3 and 2); the smaller child, 2, wins, so they swap: [2, 3, 8, 5, 9]. 8 now sits at index 2, which has no children left (index 5 is out of range for a 5-element list), so sifting stops. Returned value: 1.
  • Second heappop: index 0 (value 2) is saved to return. The last element, 9, moves into index 0, giving [9, 3, 8, 5]. 9 is compared against its children (3 and 8); 3 wins, so they swap: [3, 9, 8, 5]. 9 now sits at index 1, whose only child is index 3 (value 5); 5 is smaller, so they swap: [3, 5, 8, 9]. Returned value: 2.

Notice the pattern: every pop only ever touches one root-to-leaf path, never the whole array — that single path is why the operation is O(log n) rather than O(n).

Common Mistakes

Mistake 1: Assuming heapq is a max-heap

It’s easy to forget that heapq always pops the smallest value, not the largest.

import heapq

numbers = [3, 1, 4, 1, 5, 9]
heapq.heapify(numbers)
largest = heapq.heappop(numbers)
print(largest)

Output:

1

Despite the variable being named largest, this prints 1 — the smallest value in the list — because heapq is unconditionally a min-heap. The fix is to negate the values before pushing them, then negate the result again on the way out:

import heapq

numbers = [3, 1, 4, 1, 5, 9]
max_heap = [-value for value in numbers]
heapq.heapify(max_heap)
largest = -heapq.heappop(max_heap)
print(largest)

Output:

9

Negating twice works because the smallest negative number is always the largest original positive number.

Mistake 2: Pushing unorderable tie-breakers

When you push (priority, payload) tuples and two priorities tie, heapq falls back to comparing the payloads directly. If the payload is something without a defined ordering, like a dict, that crashes.

import heapq

heap = []
heapq.heappush(heap, (2, {'task': 'email'}))

# The second push has the same priority (2), so heapq compares the
# second tuple element to break the tie. Dicts do not support
# ordering, so this line raises:
# TypeError: '<' not supported between instances of 'dict' and 'dict'
heapq.heappush(heap, (2, {'task': 'backup'}))

Output:

TypeError: '<' not supported between instances of 'dict' and 'dict'

The fix is to add a middle tiebreaker element that is always comparable and always unique, so the payload is never reached during comparison. itertools.count() is the standard tool for this — it hands out an ever-increasing integer each time you call next() on it:

import heapq
import itertools

counter = itertools.count()
heap: list[tuple[int, int, dict[str, str]]] = []

heapq.heappush(heap, (2, next(counter), {'task': 'email'}))
heapq.heappush(heap, (2, next(counter), {'task': 'backup'}))
heapq.heappush(heap, (1, next(counter), {'task': 'alert'}))

while heap:
    priority, order, payload = heapq.heappop(heap)
    print(priority, payload['task'])

Output:

1 alert
2 email
2 backup

The counter values (0, 1, 2) are themselves unique integers, so whenever priorities tie, the comparison stops at the counter and never has to look at the dict — and because the counter only increases, it also preserves insertion order among equal priorities, which matches how the earlier string-based example happened to behave only by coincidence.

Best Practices

  • Use heapq whenever you repeatedly need ‘the smallest (or highest-priority) item next’ — it beats resorting a list after every insert, which costs O(n log n) per operation instead of O(log n).
  • Build a heap from existing data with heapq.heapify() in O(n), rather than calling heappush in a loop, which costs O(n log n) overall.
  • There is no built-in max-heap — negate the priority (or the whole value, if it’s just numbers) to flip a min-heap into a max-heap.
  • When items can tie on priority and aren’t naturally orderable, push (priority, tiebreaker, item) with an itertools.count() tiebreaker so the comparison never reaches the item itself.
  • For ‘find the top k’ problems, reach for heapq.nlargest(k, iterable) or heapq.nsmallest(k, iterable) instead of sorting the whole collection — it’s O(n log k) instead of O(n log n).
  • heapq has no direct ‘decrease priority’ or ‘remove arbitrary item’ operation. The standard workaround is lazy deletion: push a new entry with the updated priority and simply skip stale entries when they’re popped, as Dijkstra’s algorithm does above.
  • Remember that only heap[0] is guaranteed to be the minimum — the rest of the underlying list is a valid heap, not a sorted sequence, so never iterate over it expecting sorted order.

Practice Exercises

  • K Closest Points to Origin. Given a list of (x, y) points and an integer k, return the k points closest to the origin using Euclidean distance, without sorting the entire list. Hint: use heapq.nsmallest(k, points, key=...) with a key function that computes squared distance (no need for a real square root, since it doesn’t change the ordering).
  • Running Median. Design a structure that supports adding numbers one at a time and, at any point, reporting the median of every number added so far in better than O(n) per query. Hint: keep two heaps — a max-heap for the smaller half of the numbers seen so far, and a min-heap for the larger half — and rebalance their sizes after every insert so they differ by at most one element.
  • Merge k Sorted Lists. Given k already-sorted lists of integers, merge them into one sorted list using a heap so that you never compare more than k elements at a time. For input [[1, 4, 7], [2, 5], [3, 6, 8, 9]] the expected output is [1, 2, 3, 4, 5, 6, 7, 8, 9]. Hint: push each list’s first element onto the heap tagged with which list it came from, and every time you pop one, push the next element from that same list.

Summary

  • A priority queue always serves the highest-priority item next, regardless of the order items were inserted.
  • Python’s heapq module implements a binary min-heap over a plain list, where index i‘s children live at 2i + 1 and 2i + 2, and its parent at (i - 1) // 2.
  • heappush and heappop are both O(log n) because they sift an element along one root-to-leaf path; heapify builds a heap from a list in O(n); peeking heap[0] is O(1); space is O(n).
  • There is no built-in max-heap — negate values (or the priority field of a tuple) to simulate one.
  • When priorities tie, heapq compares the next tuple element — add an itertools.count() tiebreaker so it never has to compare unorderable payloads.
  • Because heapq can’t update or remove an arbitrary element, real code handles changing priorities with lazy deletion: push a fresh entry and skip stale ones on pop, as Dijkstra’s algorithm does.
  • Real-world uses include task scheduling, Dijkstra’s and A* shortest-path search, top-k and streaming problems, k-way merges, and event simulation.