Priority Queues and Min-Heaps with heapq
20 min · Intermediate
Need help? Ask a tutor on WhatsApp (opens in a new tab)A heap is a specialized complete binary tree that provides fast retrieval of the minimum element (O(1) peek). Python’s standard library provides the heapq module, which implements min-heaps directly on regular Python lists without extra object overhead.
Binary Min-Heap Array Representation
Min-heap is a complete binary tree where every parent is smaller than or equal to its children, so the smallest item is always at the root.
A min-heap satisfies the heap property: heap[0] is always the smallest element, and for every index i:
- Left child:
2*i + 1 - Right child:
2*i + 2 - Parent:
(i - 1) // 2
Essential heapq Operations
Priority queue is a collection that always removes the item with the highest priority first, not the oldest. In heapq, the smallest value has the highest priority.
The heapq module operates directly on standard Python lists.
import heapq
heap = []heapq.heappush(heap, 30)heapq.heappush(heap, 10)heapq.heappush(heap, 20)
print("Peek minimum:", heap[0]) # 10 (O(1))print("Popped:", heapq.heappop(heap)) # 10 (O(log N))print("New minimum:", heap[0]) # 20Example 1 · Priority task scheduler
import heapq
# Priority Task Scheduler (priority, task_name)tasks = []heapq.heappush(tasks, (2, "Write documentation"))heapq.heappush(tasks, (1, "Fix production outage"))heapq.heappush(tasks, (3, "Refactor CSS"))
execution_order = []while tasks: priority, task = heapq.heappop(tasks) execution_order.append(f"{task} (P{priority})")
print("Executed:", " -> ".join(execution_order))Executed: Fix production outage (P1) -> Write documentation (P2) -> Refactor CSS (P3)
What this code does
- Tuples
(priority, task_name)are compared by their first elementpriority. - Tasks execute in strict ascending priority order (1 first).
Example 2 · Equal priorities need a tiebreaker
import heapq
tickets = [(2, {"id": "T-7"}), (1, {"id": "T-9"}), (2, {"id": "T-3"})]queue = []for seq, (priority, ticket) in enumerate(tickets): heapq.heappush(queue, (priority, seq, ticket))
while queue: priority, _, ticket = heapq.heappop(queue) print(priority, ticket["id"])1 T-9 2 T-7 2 T-3
What this code does
- Without
seq, two tickets with priority 2 would make Python compare the dicts and raise TypeError. seqis unique, so it also keeps equal priorities in arrival order.
heappop() removes the root and restores the heap property in O(log N) time.
Building a Heap in Linear O(N) Time
Do not push elements one by one (O(N log N)). Use heapq.heapify() to convert an existing list in-place in linear O(N) time.
numbers = [50, 10, 40, 20, 30]heapq.heapify(numbers) # In-place O(N)print("Heapified array:", numbers) # numbers[0] is guaranteed to be 10Example 3 · Heapify, then pop in order
import heapq
prices = [49, 15, 99, 5, 30]heapq.heapify(prices)print(prices[0])print([heapq.heappop(prices) for _ in range(len(prices))])5 [5, 15, 30, 49, 99]
What this code does
heapifyrearranges the list in O(N) and puts the minimum at index 0.- Popping every item gives sorted order in O(N log N), which is how heap sort works.
Heapify is standard for initialization in coding rounds.
Simulating a Max-Heap and Finding Top-K
Python only provides min-heaps. To simulate a max-heap, negate values (-val).
# Max-heap simulationmax_heap = []for num in [15, 80, 45]: heapq.heappush(max_heap, -num)
largest = -heapq.heappop(max_heap) # 80!
# Or use convenience functions:top_2 = heapq.nlargest(2, [15, 80, 45, 99]) # [99, 80]Example 4 · Top 2 products by revenue
import heapq
products = [ {"name": "Laptop", "revenue": 90000}, {"name": "Mouse", "revenue": 12000}, {"name": "Monitor", "revenue": 45000},]best = heapq.nlargest(2, products, key=lambda p: p["revenue"])print([p["name"] for p in best])['Laptop', 'Monitor']
What this code does
keytellsnlargestwhat to compare, so dicts need no tuples.- For small k this is faster than sorting the whole list.
nlargest and nsmallest are optimized for small K.
Keeping the K Largest in a Stream
A min-heap of size k holds the k largest items seen so far. Its root is the smallest of them, so it is the one to replace.
import heapq
scores = [72, 91, 55, 88, 64, 97, 80]top3 = []for score in scores: if len(top3) < 3: heapq.heappush(top3, score) else: heapq.heappushpop(top3, score)print(sorted(top3, reverse=True))Example 5 · Running 3rd-largest score
import heapq
heap = []for score in [50, 80, 60, 90, 70]: heapq.heappush(heap, score) if len(heap) > 3: heapq.heappop(heap) if len(heap) == 3: print(f"after {score}: 3rd largest is {heap[0]}")after 60: 3rd largest is 50 after 90: 3rd largest is 60 after 70: 3rd largest is 70
What this code does
- The heap never holds more than 3 scores, so every step costs O(log 3).
- The root
heap[0]is the smallest of the top 3, which is the 3rd largest.
Each score costs O(log k) and memory stays at k items, even if the stream never ends.
Next step
Check your understanding
11 quick questions