Heaps and Priority Queues

Always know the smallest item

A binary heap is a complete binary tree stored in an array where every parent is less than or equal to its children (a min-heap), so the smallest element is always at index 0. Python's heapq module implements a min-heap on a plain list: heappush and heappop are O(log n), peeking at heap[0] is O(1), and heapify turns a list into a heap in O(n). Heaps implement priority queues (task schedulers, event simulations, Dijkstra's algorithm), top-k queries (heapq.nlargest(k, items, key=...) and nsmallest, O(n log k)), streaming medians (with two heaps) and merging sorted streams (heapq.merge). For a max-heap, push negated priorities, or (from Python 3.14) use the new max-heap functions. To prioritise objects that are not comparable, push tuples (priority, counter, item): the unique counter breaks ties and preserves insertion order. To change a priority, the usual approach is lazy deletion: push a new entry and mark the old one as removed, skipping stale entries when popping.

A task scheduler and top-k with heapq

Tuples with a tie-breaking counter, heapify and nlargest.

import heapq
from itertools import count

class TaskQueue:
    def __init__(self):
        self._heap = []
        self._counter = count()                   # tie-breaker keeps FIFO order for equal priorities

    def push(self, priority: int, task: str) -> None:
        heapq.heappush(self._heap, (priority, next(self._counter), task))

    def pop(self) -> str:
        priority, _, task = heapq.heappop(self._heap)
        return task

    def __len__(self) -> int:
        return len(self._heap)

q = TaskQueue()
q.push(2, "send newsletter")
q.push(0, "process payment")                    # lower number = higher priority
q.push(1, "generate invoice")
q.push(0, "fraud check")
while q:
    print(q.pop())   # process payment, fraud check, generate invoice, send newsletter

latencies = [120, 45, 300, 87, 950, 60, 410]
heapq.heapify(latencies)                        # O(n), in place
print(latencies[0])                             # 45: smallest at the root
print(heapq.nlargest(3, [120, 45, 300, 87, 950, 60, 410]))   # [950, 410, 300]

orders = [{"id": "o1", "total": 1200}, {"id": "o2", "total": 300}, {"id": "o3", "total": 800}]
print([o["id"] for o in heapq.nlargest(2, orders, key=lambda o: o["total"])])   # ['o1', 'o3']

heapq is a min-heap

heapq always pops the smallest item. For a max-heap of numbers, push -value and negate when popping, or use nlargest for one-off queries.

Quick check: What is the time complexity of heapq.heapify on a list of n items?

• O(n)
• O(n log n)
• O(log n)
• O(1)