Practical Priority Queues in Python with heapq

Many programs need to repeatedly choose the most important pending item rather than process items in insertion order. Schedulers pick the next deadline, graph algorithms choose the lowest-cost candidate, and streaming systems keep only the best few observations seen so far.

A sorted list can solve these problems, but maintaining full ordering is often unnecessary. Python’s heapq module provides a heap: a compact data structure that keeps one extreme element immediately available while doing only enough work to preserve that property.

The most important design point is that heapq is not a general sorted container. It is a tool for workloads where repeated access to the smallest item, partial ordering, or bounded selection matters more than maintaining a completely sorted sequence.

Understand the heap invariant

For the traditional heapq min-heap operations, the smallest item is always at index 0.

import heapq

values = [9, 3, 7, 1, 5]
heapq.heapify(values)

print(values[0])

The result is 1.

The rest of values is not guaranteed to be globally sorted. The heap only maintains the relationship needed to ensure that every parent is no greater than its children.

That distinction prevents a common mistake:

heapq.heapify(values)

# Wrong assumption: values is now sorted.
print(values)

If you need all elements in sorted order, use sorted(). If you need to repeatedly remove the smallest pending element, a heap is a better fit.

Build a heap efficiently

An empty list can become a heap by pushing items:

import heapq

heap = []

for value in [9, 3, 7, 1, 5]:
    heapq.heappush(heap, value)

When all initial values are already available, heapify() is usually the clearer operation:

import heapq

heap = [9, 3, 7, 1, 5]
heapq.heapify(heap)

heapify() transforms the list in place in linear time. It does not return a new heap.

The list remains an ordinary mutable list, but after heap construction, mutations that affect its contents should go through heap operations unless you deliberately restore the invariant afterward. Calling append(), removing an arbitrary element, or assigning a different value to an internal position can invalidate the heap.

Push and pop the next item

heappush() inserts an item while preserving the heap invariant. heappop() removes and returns the smallest item:

import heapq

heap = []

heapq.heappush(heap, 30)
heapq.heappush(heap, 10)
heapq.heappush(heap, 20)

while heap:
    print(heapq.heappop(heap))

The values are produced as:

10
20
30

Each push or pop adjusts only the path needed to restore the heap property rather than re-sorting the whole list.

Calling heappop() on an empty heap raises IndexError. If emptiness is expected, check the heap before popping or structure the surrounding control flow so a pop is attempted only when work exists.

Peek without removing

Because the smallest item is at the root, peeking is simply:

if heap:
    next_item = heap[0]

Do not use min(heap) for this. The heap already maintains the minimum at index 0.

Store priorities with the payload

Priority queues usually need to associate a priority with some application object.

Tuples are convenient because Python compares them lexicographically:

import heapq

queue = []

heapq.heappush(queue, (20, "send report"))
heapq.heappush(queue, (10, "refresh token"))
heapq.heappush(queue, (30, "archive logs"))

priority, task = heapq.heappop(queue)

print(priority)
print(task)

The smallest priority value comes out first.

This representation is appropriate when lower numeric values mean higher scheduling priority. If your domain uses the opposite convention, convert the domain priority deliberately rather than letting the representation remain ambiguous.

Break ties without comparing tasks

A two-item tuple can fail when two priorities are equal and the payload objects are not orderable.

For example:

class Task:
    def __init__(self, name):
        self.name = name

Two entries such as (10, Task("a")) and (10, Task("b")) first compare their equal priorities, then Python tries to compare the Task instances.

A standard solution is to add a monotonically increasing sequence number:

import heapq
import itertools


class Task:
    def __init__(self, name):
        self.name = name


counter = itertools.count()
queue = []

heapq.heappush(queue, (10, next(counter), Task("first")))
heapq.heappush(queue, (10, next(counter), Task("second")))

priority, sequence, task = heapq.heappop(queue)

print(task.name)

The sequence value has two benefits. It prevents comparison from reaching the payload, and it preserves insertion order among entries with equal priorities.

This pattern is useful even when current payload objects happen to be comparable. It makes tie behavior explicit instead of depending on an unrelated ordering defined by the payload type.

Do not edit priorities in place

Changing an element already inside the list can break the heap invariant:

# Do not do this to an arbitrary heap entry.
queue[5] = (1, sequence, task)

Even if you know where an item is stored, assigning a new priority does not move it to the correct location.

For a small queue, rebuilding with heapify() after a batch of deliberate changes can be reasonable. For a long-lived priority queue with frequent updates or cancellations, a common design is lazy deletion.

Handle updates with lazy deletion

Instead of searching the heap and removing an old entry, keep a mapping to the currently active entry and mark replaced entries as removed.

import heapq
import itertools

REMOVED = object()
counter = itertools.count()

heap = []
entries = {}


def add(task_id, priority):
    if task_id in entries:
        cancel(task_id)

    entry = [priority, next(counter), task_id]
    entries[task_id] = entry
    heapq.heappush(heap, entry)


def cancel(task_id):
    entry = entries.pop(task_id)
    entry[2] = REMOVED


def pop_next():
    while heap:
        priority, sequence, task_id = heapq.heappop(heap)

        if task_id is not REMOVED:
            del entries[task_id]
            return task_id, priority

    raise KeyError("priority queue is empty")

Updating a task means canceling its current entry and pushing a new one.

Old entries remain in the heap until they reach the root and are discarded. This avoids arbitrary heap deletion, but it creates a trade-off: a workload with many updates and few pops can accumulate stale entries.

If that retained memory matters, periodically rebuild the heap from active entries:

heap[:] = [entry for entry in heap if entry[2] is not REMOVED]
heapq.heapify(heap)

Rebuilding has a cost, so trigger it based on an application-specific threshold rather than after every cancellation.

Keep identity and equality separate

The example uses a unique sentinel object and checks it with is. That avoids confusing a legitimate task identifier with the removed marker.

If task IDs come from users or external systems, do not reserve an ordinary string such as "<removed>" and assume it can never collide with real data.

Choose heappushpop and heapreplace carefully

heapq provides two combined operations that look similar but have different semantics.

heappushpop(heap, item) conceptually pushes the new item and then removes the smallest. The combined implementation is more efficient than separate heappush() and heappop() calls.

import heapq

heap = [10, 20, 30]
heapq.heapify(heap)

removed = heapq.heappushpop(heap, 25)

print(removed)
print(heap)

The removed value is 10, leaving the larger values in the heap.

heapreplace(heap, item) instead removes the current smallest item and then inserts the new item. The heap must be non-empty.

The difference matters when the new item is smaller than the current root. With heappushpop(), that new small item can be returned immediately, leaving the old heap unchanged. With heapreplace(), the old root is always returned and the new item remains in the heap.

Choose the operation based on the required result, not merely because both keep the heap size unchanged.

Keep the largest k values with a min-heap

A min-heap is useful even when the goal is to retain the largest values seen in a stream.

Keep at most k items. The root then represents the smallest value currently accepted into the top set:

import heapq


def largest_k(values, k):
    if k <= 0:
        return []

    heap = []

    for value in values:
        if len(heap) < k:
            heapq.heappush(heap, value)
        elif value > heap[0]:
            heapq.heapreplace(heap, value)

    return sorted(heap, reverse=True)


print(largest_k([8, 1, 6, 3, 9, 2, 7], 3))

The result is:

[9, 8, 7]

Once the heap is full, values no larger than heap[0] cannot belong to the largest k, so they can be ignored.

The standard library already provides heapq.nlargest() and heapq.nsmallest() for this general task. Prefer those functions when their interface matches the requirement. A custom bounded heap becomes useful when selection is integrated into a longer-lived streaming process or when the application needs to inspect or update state incrementally.

Do not assume a heap is the best partial-sort strategy at every size

nlargest() and nsmallest() are intended for relatively small selections. When n is large relative to the input, a full sorted() operation may be more appropriate.

For the special case of one item, max() or min() communicates the requirement directly.

Performance decisions here depend on input size, key cost, and how much of the result is needed. Benchmark the real workload when this is a hot path.

Merge already sorted streams lazily

heapq.merge() combines multiple sorted inputs into one sorted iterator:

import heapq

service_a = [1, 4, 9]
service_b = [2, 3, 10]
service_c = [5, 6, 7]

for value in heapq.merge(service_a, service_b, service_c):
    print(value)

The result is globally ordered without first concatenating every input and sorting the combined collection.

This is particularly useful when inputs are already sorted and may be large or streamed incrementally.

The precondition matters: each input must already be sorted according to the same ordering expected by merge(). If an input is unsorted, merge() does not repair it.

merge() also accepts key and reverse keyword arguments. When reverse=True, the inputs must themselves be sorted in descending order.

Distinguish heapq from a thread-safe queue

A list managed with heapq does not provide synchronization for application-level producer and consumer coordination.

For multi-threaded code that needs a synchronized priority queue, the standard library provides queue.PriorityQueue, which uses locking for multi-producer, multi-consumer use.

That abstraction has different goals from using heapq directly. heapq is appropriate when the surrounding code already owns synchronization, runs in one thread, or needs custom queue behavior. PriorityQueue is appropriate when blocking and thread-safe queue operations are part of the requirement.

Do not assume that a correct heap invariant also solves concurrency.

Avoid version-dependent max-heap assumptions

Recent Python versions include dedicated max-heap helpers, but code that needs to run across a broad range of Python versions should not assume those APIs are present.

For portable code, first ask whether a max-heap is actually necessary. Many “keep the largest values” problems are naturally solved by a min-heap whose root is the current cutoff.

When a domain truly requires repeated maximum extraction and the supported Python baseline is known, choose an implementation appropriate for that baseline. Avoid scattering version-specific heap behavior through business logic.

Common pitfalls

Treating the internal list as sorted

Only the root is guaranteed to be the smallest element of a min-heap. Use repeated pops or sorted() when complete ordering is required.

Mutating heap entries arbitrarily

Changing priorities or appending directly can violate the invariant. Use heap operations or deliberately rebuild with heapify().

Ignoring tie behavior

Equal priorities can cause Python to compare payloads. Add a sequence number when payloads should not participate in ordering.

Using negated priorities without checking the domain

Negating numeric priorities is a familiar way to emulate maximum-first behavior, but it couples representation to numeric values and can make mixed conventions hard to read. Document the convention and keep it at the queue boundary.

Forgetting stale entries in lazy deletion

Lazy deletion makes updates practical, but canceled entries still consume memory until removed or the heap is rebuilt.

Using a heap when sorting once is simpler

If all data is already in memory and the program only needs one fully ordered result, sorted() is usually clearer. A heap earns its complexity when operations are incremental or only partial ordering is needed.

Pick the data structure for the access pattern

A priority queue is valuable because it avoids maintaining order that the program never uses.

Use a heap when the workload repeatedly needs the smallest pending item, incrementally maintains a bounded best set, or merges already sorted streams. Add an explicit tie-breaker when payload ordering is irrelevant, and use lazy deletion carefully when priorities can change.

Use a full sort when the whole result must be ordered, min() or max() when only one extreme is needed, and a synchronized queue when threads need coordinated access.

heapq stays simple when its invariant remains visible in the design: the root is special, the rest is only partially ordered, and every mutation must preserve that relationship.