Python Priority Queues with heapq

Intermediate

A deque is useful when a program serves items from the ends in arrival order. A priority queue solves a different problem: each item carries a priority, and the item with the smallest priority should be selected first. Python’s heapq module supports this pattern using an ordinary list arranged as a min-heap.

Understand the min-heap invariant

A min-heap is not a fully sorted list. Its invariant is that each parent is less than or equal to its children, so the smallest value is always at index 0. The remaining values can appear in an order that is not globally sorted. Treat the list as a heap and use heapq operations to keep that invariant intact.

import heapq

values = [9, 4, 7, 1]
heapq.heapify(values)
print(values[0])?

Build a heap and take the smallest item

Start with an empty list or call heapify() on an existing list. heapify() rearranges that list in place and takes linear time. heappush() adds an item while restoring the invariant; heappop() removes and returns the smallest item. heappop() raises IndexError if the heap is empty. Use heap[0] to inspect the smallest item without removing it.

import heapq

heap = [2, 5, 8]
heapq.heappush(heap, 4)
print(heapq.heappop(heap))?

Represent tasks with priorities

For a priority queue, store entries such as (priority, sequence_number, task). The smallest priority is removed first. A monotonically increasing sequence number breaks ties in arrival order and prevents tuple comparison from reaching task objects that may not support ordering. The sequence number is a tie-breaker, not part of the task’s business priority.

import heapq
from itertools import count

sequence = count()
heap = []
heapq.heappush(heap, (1, next(sequence), "first"))
heapq.heappush(heap, (1, next(sequence), "second"))
print(heapq.heappop(heap)[2])?

Choose between heappushpop() and heapreplace()

Both operations combine a push and a pop, but they do not mean the same thing. heappushpop(heap, item) returns the smaller of the incoming item and the heap’s current minimum, leaving the larger in the heap. heapreplace(heap, item) always removes and returns the old minimum, then inserts the new item; the heap must already be non-empty, and the returned old minimum can be greater than the incoming value.

import heapq

heap = [2, 5, 7]
removed = heapq.heapreplace(heap, 1)
print(removed, heap[0])?

Choose the right queue abstraction

Use heapq when each item’s priority determines what should be removed next, for example selecting the next scheduled job or repeatedly keeping the smallest values. Use deque for efficient operations at either end in FIFO/LIFO patterns. Use queue.PriorityQueue when multiple threads need the queue’s locking and coordination behavior; heapq itself does not provide thread synchronization.

import heapq

queue = []
for priority, name in [(5, "later"), (1, "urgent")]:
    heapq.heappush(queue, (priority, name))
print(heapq.heappop(queue)[1])?