heapq
The heapq module turns a plain Python list into a binary min-heap, giving you O(log n) push/pop for the smallest item -- the engine behind priority queues and top-k problems.
A heap is a tree-shaped data structure that keeps one operation extremely cheap: finding the smallest (or largest) item. Python's heapq module doesn't introduce a new type -- it provides functions that treat an ordinary list as a binary min-heap, maintaining the heap invariant as you push and pop.
What Is a Heap
A heap is a binary tree stored compactly in an array (a Python list here), where each parent is less than or equal to its children (a min-heap) -- so the smallest element is always at the root. heapq operates on plain lists directly: there is no special Heap class, just functions that keep a list in heap order.
heapify() rearranges the existing list in place in O(n) time; the resulting order is not the same as sorted(), only the heap property (parent <= children) is guaranteed.
import heapq
nums = [5, 1, 8, 3, 9, 2]
heapq.heapify(nums) # rearranges in place into heap order, O(n)
print(nums) # e.g. [1, 3, 2, 5, 9, 8] -- not fully sorted, but heap-ordered
print(nums[0]) # 1 -- the smallest item is always at index 0The Min-Heap Invariant
heapq always implements a min-heap: for every parent at index i, its children at indices 2i+1 and 2i+2 are each greater than or equal to it. This invariant guarantees heap[0] is always the minimum, and every push/pop operation restores the invariant in O(log n) by moving the changed element up or down the tree ('sift up' / 'sift down').
Heap invariant (min-heap):
1
/ \
3 2
/ \ /
5 9 8
Every parent <= its children -- but siblings are NOT ordered relative to each other.Pushing and Popping
heappush(heap, item) adds item and restores the invariant in O(log n). heappop(heap) removes and returns the smallest item, then restores the invariant, also in O(log n). Both mutate the list in place; there is no separate heap object to manage.
Repeatedly popping from a heap always yields items in ascending order -- this is the basis of heapsort, though Python's sorted() uses a different, typically faster algorithm (Timsort) for general sorting.
import heapq
heap: list[int] = []
for n in [5, 1, 8, 3, 9]:
heapq.heappush(heap, n)
print(heap[0]) # 1 -- peek at the minimum without removing it
order = []
while heap:
order.append(heapq.heappop(heap))
print(order) # [1, 3, 5, 8, 9] -- popping repeatedly yields sorted orderheappushpop and heapreplace
heapq offers two combined operations that are each more efficient than calling push then pop separately: heappushpop(heap, item) pushes item then pops and returns the smallest (skipping the push entirely if item is already the smallest); heapreplace(heap, item) pops the smallest first, then pushes item -- useful for maintaining a fixed-size heap.
heap = [2, 4, 6]
heapq.heapify(heap)
smallest = heapq.heapreplace(heap, 1) # pop 2, then push 1
print(smallest, heap) # 2 [1, 4, 6]Top-K Problems: nlargest and nsmallest
heapq.nlargest(n, iterable, key=None) and heapq.nsmallest(n, iterable, key=None) return the n largest/smallest items as a sorted list, without fully sorting the whole iterable. For small n relative to the input size, these are more efficient than sorted(iterable)[:n] or [-n:], since they avoid a full O(n log n) sort.
For n=1, min()/max() are simpler and just as fast; nlargest/nsmallest earn their keep specifically for 'top-k where 1 < n << len(iterable)'.
scores = [55, 92, 78, 61, 99, 40, 88]
print(heapq.nlargest(3, scores)) # [99, 92, 88]
print(heapq.nsmallest(2, scores)) # [40, 55]
records = [('Ana', 88), ('Bo', 95), ('Cy', 72)]
print(heapq.nlargest(2, records, key=lambda r: r[1])) # top 2 by scoreMax-Heap via Negation
heapq only implements a min-heap directly. To use it as a max-heap, negate values on the way in and out (or push (-priority, item) tuples), since -min(-x for x in xs) == max(xs).
This trick works because negating reverses order: the smallest negated value corresponds to the largest original value.
nums = [5, 1, 8, 3]
max_heap = [-n for n in nums]
heapq.heapify(max_heap)
largest = -heapq.heappop(max_heap)
print(largest) # 8 -- the true maximum, recovered by re-negatingPriority Queues with heapq
A common use of heapq is a priority queue: push (priority, item) tuples, and heappop always returns the lowest-priority item first. Python compares tuples element by element, so ties on priority fall through to comparing item -- which raises TypeError if item isn't orderable (e.g. two dicts). Add a tie-breaking counter to avoid comparing the items themselves.
The counter guarantees no two tuples ever compare equal on the first two elements, so the third element (the task, which may not support <) is never compared.
import heapq
import itertools
counter = itertools.count() # tie-breaker: always increasing, always comparable
pq: list[tuple[int, int, str]] = []
for priority, task in [(3, 'low'), (1, 'urgent'), (2, 'medium')]:
heapq.heappush(pq, (priority, next(counter), task))
while pq:
priority, _, task = heapq.heappop(pq)
print(priority, task)
# 1 urgent
# 2 medium
# 3 lowCommon Mistakes
Two mistakes dominate heapq usage: assuming a heapified list is fully sorted (only the root is guaranteed minimal), and pushing (priority, payload) tuples where payload can tie and isn't comparable, causing a TypeError deep in a tie-break comparison.
Complete Example: Merging Sorted Streams
heapq.merge() lazily merges any number of already-sorted iterables into one sorted iterator using a heap internally -- useful for combining sorted log files or sorted query results without loading everything into memory at once.
import heapq
a = [1, 4, 7, 10]
b = [2, 3, 8, 9]
c = [0, 5, 6]
merged = list(heapq.merge(a, b, c))
print(merged) # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]Key Takeaways
heapq operates on plain Python lists in place -- there is no dedicated Heap class.
It implements a min-heap: heap[0] is always the smallest element, but the rest of the list is only partially ordered.
heappush/heappop are each O(log n); heapify() builds a heap from an existing list in O(n).
nlargest(n, ...) / nsmallest(n, ...) are the efficient way to get the top-k items without a full sort.
Simulate a max-heap by negating values, since heapq only provides min-heap behavior directly.
Use an increasing tie-breaker in priority-queue tuples to avoid comparing non-orderable payloads.
heapq.heapify(nums) print(nums) # assuming this is now fully sorted heapq.heapify(nums) print(nums[0]) # only the minimum (index 0) is guaranteed heapify() only guarantees the heap invariant (each parent <= its children), not a fully sorted list -- only heap[0] is guaranteed to be the smallest element.
heapq.heappush(pq, (priority, task_dict)) # TypeError if two priorities tie and dicts aren't comparable heapq.heappush(pq, (priority, next(counter), task_dict))
Tuple comparison falls through to the second element on a tie; adding an always-increasing counter as the second element guarantees no comparison ever reaches the (possibly non-orderable) payload.
largest = heapq.heappop(nums) # thinking heappop always removes the largest largest = -heapq.heappop(max_heap) # after building the heap with negated values heapq is always a min-heap; heappop always returns the smallest element. To retrieve the largest, negate values on the way in (and re-negate on the way out).
What guarantee does the heap invariant give you about a heapified list, and what does it not guarantee?
It guarantees index 0 holds the smallest element (every parent is <= its children), but it does not guarantee the rest of the list is fully sorted -- siblings can appear in either order relative to each other.
Why is heapq.nlargest(3, big_list) usually faster than sorted(big_list)[-3:] for a large list?
nlargest only does the work needed to find the top 3 (roughly O(n log 3)), while sorted() fully orders every element (O(n log n)) before slicing -- most of that sorting work is wasted when only a few items are needed.
How do you use heapq to build a max-heap, given that it only implements a min-heap directly?
Negate every value going in (push -x instead of x) and negate again on the way out (-heappop(heap)); since the smallest negated value corresponds to the largest original value, this recovers max-heap behavior.
Why might heapq.heappush(pq, (priority, task)) raise a TypeError, and how do you prevent it?
If two entries have equal priority, tuple comparison falls through to comparing task; if task isn't orderable (e.g. a dict), Python raises TypeError. Insert an always-increasing counter as a middle tuple element so ties are broken there instead of ever comparing task.
What is the time complexity of heapq.heapify() versus calling heappush() n times on an empty list?
heapify() builds a heap from an existing list in O(n) total; pushing n items one at a time is O(n log n) total, since each individual heappush is O(log n) -- heapify is the faster way to build a heap from data you already have.
What does heapq.merge() do, and why is it more memory-efficient than concatenating and sorting several lists?
It lazily merges multiple already-sorted iterables into one sorted iterator using a heap internally, producing items on demand instead of materializing and sorting a combined list -- useful for large or streaming sorted inputs.
Write a function that returns the k smallest values from a list using heapq, without sorting the entire list.
import heapq def k_smallest(values: list[int], k: int) -> list[int]: return heapq.nsmallest(k, values) print(k_smallest([9, 3, 7, 1, 8, 2], 3)) heapq.nsmallest(k, values) finds the k smallest items using heap-based selection, which is more efficient than sorted(values)[:k] for k much smaller than len(values) -- result: [1, 2, 3].
What does this print?
0 -- pushing 0 onto the heap restores the min-heap invariant, and since 0 is smaller than every existing element, it becomes the new root at index 0.
Write a simple task scheduler that processes tasks in priority order (lower number = higher priority) using heapq, breaking ties by insertion order.
import heapq import itertools def run_in_priority_order(tasks: list[tuple[int, str]]) -> list[str]: counter = itertools.count() pq = [(priority, next(counter), name) for priority, name in tasks] heapq.heapify(pq) order = [] while pq: _, _, name = heapq.heappop(pq) order.append(name) return order print(run_in_priority_order([(2, 'medium'), (1, 'urgent'), (2, 'also-medium')]))
The counter (an ever-increasing itertools.count()) is included as the tiebreaker element, so two tasks with equal priority are ordered by insertion order instead of risking a comparison between the task name strings -- result: ['urgent', 'medium', 'also-medium'].
import heapq
h = [5, 1, 8]
heapq.heapify(h)
heapq.heappush(h, 0)
print(h[0])