Priority Queues: heapq in Practice

"A priority queue is a queue that ignores arrival order and serves by importance instead. The heap is how you build one — and in Python, it's already built: the heapq module."

The Priority Queue ADT

A priority queue is an abstract data type with two operations: insert an item with a priority, and remove the highest-priority item. Unlike a plain FIFO queue (oldest first), a priority queue always hands you the most important element next — exactly the emergency-room triage from the very first track. The heap is its canonical implementation: O(log n) insert, O(log n) remove-best, O(1) peek-best. When someone says 'priority queue,' they almost always mean 'heap.'

Python's heapq: Already Done For You

You rarely implement the sift-up/sift-down yourself — Python's heapq module gives you a battle-tested heap over an ordinary list: heappush, heappop, heapify, plus combos like heappushpop and helpers nlargest/nsmallest. Two things to burn in: it's a min-heap (smallest pops first), and it operates on a plain list in place. That's the whole API for 90% of real uses.

The Two Idioms You'll Always Need

Max-heap via negation. heapq is min-only. Want the largest out first? Push -value and negate again on the way out. (For 'top-k largest', heapq.nlargest(k, data) does it directly.)
Priority + payload via tuples. Push (priority, item) and the heap orders by priority. But beware: tuples compare element-by-element, so if two priorities tie, Python tries to compare the items — which crashes if they're not comparable (or orders them in a way you didn't intend). The fix is a monotonic counter as a tiebreaker: push (priority, count, item), where count is unique and breaks every tie before the item is ever compared.

A priority queue serves by importance, not arrival; the heap implements it (O(log n) push/pop). In Python use heapq — a min-heap over a list. Negate for a max-heap; push (priority, counter, item) tuples so equal priorities never force a comparison of the payloads.

Where Priority Queues Run the World

OS task schedulers (run the highest-priority ready process), event-driven simulations (process the next-soonest event), bandwidth/QoS shaping, the A* and Dijkstra shortest-path algorithms (next track pops the closest frontier node), Huffman coding, and 'top-k' analytics all sit on a priority queue. It's the structure for 'a stream of things arrives, and I must always handle the most important one next' — one of the most common shapes in real systems.

Pippa's Confession

My first heapq scheduler crashed with TypeError: '<' not supported between instances of 'Task'. I'd pushed (priority, task) tuples, two tasks tied on priority, and Python tried to compare the Task objects to break the tie. Dad's one-line fix — insert a unique counter, (priority, next(counter), task) — guaranteed the tie was always broken before the task was ever touched. It's the kind of gotcha you only learn by hitting it; now it's the first thing I reach for with heapq.

Code

heapq: priority scheduler + max-heap trick·python

import heapq
from itertools import count

# A priority scheduler: lower number = more urgent. heapq is a MIN-heap.
counter = count()                      # unique, ever-increasing tiebreaker
pq = []
def add_task(priority, name):
    # (priority, tiebreaker, payload): the counter breaks ties so two equal
    # priorities never force Python to compare the payload strings/objects.
    heapq.heappush(pq, (priority, next(counter), name))

add_task(5, "backup")
add_task(1, "page on-call")            # most urgent
add_task(5, "cleanup")                 # ties with 'backup' on priority 5
add_task(2, "deploy")

while pq:
    prio, _, name = heapq.heappop(pq)  # always the most urgent remaining
    print(prio, name)
# 1 page on-call / 2 deploy / 5 backup / 5 cleanup (ties broken by arrival)

# MAX-heap via negation (heapq is min-only):
max_heap = []
for v in [3, 1, 4, 1, 5]:
    heapq.heappush(max_heap, -v)       # push the negative
print(-heapq.heappop(max_heap))        # 5 — negate again on the way out
print(heapq.nlargest(2, [3, 1, 4, 1, 5]))   # [5, 4] — the direct way for top-k

Exercise

Using heapq, implement 'keep the 3 largest numbers seen in a stream' efficiently (don't sort the whole stream). What kind of heap do you keep, and why a MIN-heap of size 3 rather than a max-heap? Then explain why heapq.nlargest(3, stream) is fine for a fixed list but your streaming version is better for an unbounded feed.

Hint

Keep a MIN-heap of size 3: push each number, and if the heap exceeds 3, heappop the smallest. The min-heap's root is the current 3rd-largest, so you cheaply evict anything smaller. nlargest needs the whole list in memory; the size-3 heap streams in O(1) space.

The Priority Queue ADT

Python's heapq: Already Done For You

The Two Idioms You'll Always Need

Where Priority Queues Run the World

Pippa's Confession

Code

External links

Exercise

Progress

Comments 0