Heaps and Priority Queues
Heaps and Priority Queues
A binary heap is a complete binary tree stored inside a standard contiguous list. A priority queue leverages this heap to retrieve the element with highest priority (smallest or largest value) in \(O(1)\) time, while insertions and deletions run in \(O(\log n)\) time. In Python, the boring default is the standard library’s heapq module, which implements an ultra-fast min-heap in C.
Mental model
A binary min-heap guarantees that every parent node is smaller than or equal to its children:
Tree view: List index view:
[ 10 (idx:0) ] +----+----+----+----+----+----+
/ \ | 10 | 14 | 19 | 26 | 31 | 42 |
[ 14 (1) ] [ 19 (2) ] +----+----+----+----+----+----+
/ \ / 0 1 2 3 4 5
[ 26(3)] [31(4)][42(5)]Parent and child index arithmetic: - \(\text{Parent}(i) = (i - 1) // 2\) - \(\text{Left}(i) = 2i + 1\) - \(\text{Right}(i) = 2i + 2\)
Operations: - heappush(heap, item) (Sift-Up): Appends to the list and bubbles the item up to restore heap order. - heappop(heap) (Sift-Down): Swaps the root with the last element, pops the last element, and sifts the new root down. - heapify(list): Transforms an arbitrary unsorted list into a valid min-heap in-place in \(O(n)\) linear time.
Worked examples
Case 1: Min-Heap from scratch using index math
Save as heap_scratch.py. Build a complete binary min-heap from first principles.
# heap_scratch.py
class MinHeap[T]:
def __init__(self) -> None:
self._items: list[T] = []
def push(self, val: T) -> None:
self._items.append(val)
self._sift_up(len(self._items) - 1)
def pop(self) -> T | None:
if not self._items:
return None
min_val = self._items[0]
last_val = self._items.pop()
if self._items:
self._items[0] = last_val
self._sift_down(0)
return min_val
def _sift_up(self, idx: int) -> None:
while idx > 0:
parent = (idx - 1) // 2
if self._items[idx] < self._items[parent]:
self._items[idx], self._items[parent] = (
self._items[parent],
self._items[idx],
)
idx = parent
else:
break
def _sift_down(self, idx: int) -> None:
n = len(self._items)
while True:
smallest = idx
left = 2 * idx + 1
right = 2 * idx + 2
if left < n and self._items[left] < self._items[smallest]:
smallest = left
if right < n and self._items[right] < self._items[smallest]:
smallest = right
if smallest != idx:
self._items[idx], self._items[smallest] = (
self._items[smallest],
self._items[idx],
)
idx = smallest
else:
break
def __len__(self) -> int:
return len(self._items)
def main() -> None:
h: MinHeap[int] = MinHeap()
scores = [42, 10, 31, 14, 26, 19]
print(f"Pushing values: {scores}")
for s in scores:
h.push(s)
print("Popping values in ascending priority order: ", end="")
while len(h) > 0:
val = h.pop()
print(f"{val} ", end="")
print()
if __name__ == "__main__":
main()Run:
uv run python heap_scratch.pyOutput:
Pushing values: [42, 10, 31, 14, 26, 19]
Popping values in ascending priority order: 10 14 19 26 31 42 Case 2: Priority Queue with standard library heapq
Save as desk_priority_queue.py. Manage urgent desk maintenance orders using Python’s standard heapq module and tie-breaking counters.
# desk_priority_queue.py
import heapq
from dataclasses import dataclass, field
@dataclass(order=True)
class PrioritizedTicket:
priority: int # 1: Emergency, 2: High, 3: Normal, 4: Low
counter: int # Secondary tie-breaker for identical priorities
item: "Ticket" = field(compare=False)
@dataclass(slots=True)
class Ticket:
id: int
title: str
class DeskPriorityQueue:
def __init__(self) -> None:
self._heap: list[PrioritizedTicket] = []
self._counter = 0
def push(self, ticket: Ticket, priority: int) -> None:
# counter ensures FIFO tie-breaking for equal priority
self._counter += 1
entry = PrioritizedTicket(
priority=priority, counter=self._counter, item=ticket
)
heapq.heappush(self._heap, entry)
def pop(self) -> Ticket | None:
if not self._heap:
return None
return heapq.heappop(self._heap).item
def __len__(self) -> int:
return len(self._heap)
def main() -> None:
pq = DeskPriorityQueue()
pq.push(Ticket(id=101, title="Restock printer paper"), priority=4)
pq.push(Ticket(id=102, title="Point-of-Sale network outage"), priority=1)
pq.push(Ticket(id=103, title="VIP Guest Check-in"), priority=2)
pq.push(Ticket(id=104, title="Refill water pitcher"), priority=4)
print("Dispatching tickets by priority:")
while len(pq) > 0:
ticket = pq.pop()
assert ticket is not None
print(f" Dispatching Ticket #{ticket.id}: {ticket.title}")
if __name__ == "__main__":
main()Run:
uv run python desk_priority_queue.pyOutput:
Dispatching tickets by priority:
Dispatching Ticket #102: Point-of-Sale network outage
Dispatching Ticket #103: VIP Guest Check-in
Dispatching Ticket #101: Restock printer paper
Dispatching Ticket #104: Refill water pitcherThe trap
When storing tuples (priority, task) in heapq, Python compares the second tuple element (task) if priorities are equal. If task is a custom class or dict that does not implement < comparison (__lt__), Python raises TypeError: '<' not supported between instances of 'Ticket' and 'Ticket'.
Always use a tie-breaker counter (priority, count, task) or a @dataclass(order=True) with field(compare=False) on the task object.
The boring rule
- Use Python’s built-in
heapqmodule for all production priority queues. It is implemented in C and runs in a fraction of the time of Python-level heaps. heapqis strictly a Min-Heap. To implement a Max-Heap, store negative priorities (-priority) or invert comparison keys.- Always include a sequence counter for tie-breaking to avoid
TypeErroron custom objects with matching priorities. - Use
heapq.heapify(lst)to turn an unsorted list into a heap in \(O(n)\) time rather than callingheappush\(n\) times in \(O(n \log n)\).
Try this
- Use
heapq.nlargest(3, orders, key=lambda o: o.price)to find the 3 most expensive orders from a list of 10,000 orders. - Implement a streaming median finder that uses two heaps: a max-heap for the lower half and a min-heap for the upper half.