heapq

The heapq module provides an implementation of the heap queue algorithm, also known as the priority queue algorithm. A heap is a binary tree where every parent node has a value less than or equal to any of its children (min-heap). The smallest element is always at the root (index 0), making it efficient to access the minimum value in O(1) time while insertions and deletions take O(log n).

Python’s heapq implements min-heaps using lists where heap[k] <= heap[2*k+1] and heap[k] <= heap[2*k+2]. This makes heap[0] the smallest item. The module is useful for priority queues, task scheduling, finding k smallest/largest elements, and merging sorted streams.

Syntax

import heapq

# Create a heap from a list
heap = [3, 1, 4, 1, 5, 9, 2, 6]
heapq.heapify(heap)

# Push and pop
heapq.heappush(heap, 0)
smallest = heapq.heappop(heap)

# Get k smallest/largest
smallest_3 = heapq.nsmallest(3, data)
largest_5 = heapq.nlargest(5, data)

The code above demonstrates the three most common patterns you will use with this module: converting a plain list into a heap with heapify, adding new values via heappush, and extracting the minimum with heappop. The functions nsmallest and nlargest are convenience wrappers that use an internal heap to find the k smallest or largest elements from any iterable without requiring you to manage the heap yourself. Each of these functions is detailed in its own section below, with real examples that show the full range of practical use cases.

heappush()

Pushes an item onto the heap while maintaining the heap invariant. The heap size increases by one.

Signature

heapq.heappush(heap, item)

Parameters

Parameter	Type	Default	Description
heap	list	—	The list to push the item onto (must be a valid heap)
item	any	—	The item to push onto the heap

Examples

Basic push operation

import heapq

# Start with an empty heap
heap = []

# Push items one by one
heapq.heappush(heap, 5)
heapq.heappush(heap, 2)
heapq.heappush(heap, 8)
heapq.heappush(heap, 2)

print(heap)
# [2, 2, 8, 5]

# The smallest is always at index 0
print(heap[0])
# 2

The example above shows the simplest use case where each heap entry is a plain number. In practice, priority queues almost always carry associated data alongside the priority value itself. The next example demonstrates the standard Python convention for attaching payloads to priorities using tuples.

Using with tuples for priority queues

import heapq

# Priority queue: (priority, task)
tasks = []
heapq.heappush(tasks, (3, 'write documentation'))
heapq.heappush(tasks, (1, 'fix critical bug'))
heapq.heappush(tasks, (2, 'code review'))

# Pop returns highest priority (lowest number) first
while tasks:
    priority, task = heapq.heappop(tasks)
    print(f"Priority {priority}: {task}")

# Output:
# Priority 1: fix critical bug
# Priority 2: code review
# Priority 3: write documentation

The tuple-based priority queue pattern shown above is the standard approach in Python for scheduling tasks by importance. The heappush examples covered adding items to a heap one at a time. The natural counterpart operation is removing items from the heap, which always returns the current smallest element while preserving the heap invariant for the remaining items. The heappop function, detailed next, handles this extraction.

heappop()

Pops and returns the smallest item from the heap. Raises IndexError if the heap is empty.

Signature

heapq.heappop(heap)

The heappop function removes and returns the smallest item from the heap, then restructures the remaining elements so the heap property continues to hold. This is the core extraction mechanism that makes heaps useful for priority queues and sorted access patterns. The following examples show how heappop works on its own and how it interacts with heapify when you need to drain a heap in sorted order.

Parameters

Parameter	Type	Default	Description
heap	list	—	The heap to pop from (must be a valid heap)

Examples

Basic pop operation

import heapq

heap = [1, 3, 5, 7, 9, 2, 4, 6, 8, 0]
heapq.heapify(heap)

# Pop all items in sorted order
while heap:
    item = heapq.heappop(heap)
    print(item, end=' ')
# 0 1 2 3 4 5 6 7 8 9

Draining a heap with repeated heappop calls produces items in ascending order, effectively implementing heapsort with O(n log n) complexity. Sometimes you only need to inspect the minimum value without removing it from the collection. The next snippet shows how to peek at the root element using direct list indexing while leaving the heap intact for further operations.

Peeking without removing

import heapq

heap = [5, 2, 8, 1, 9]
heapq.heapify(heap)

# Access smallest without popping
print(heap[0])
# 1

# Heap remains unchanged
print(len(heap))
# 5

The heappop examples demonstrated how to extract items while maintaining the heap property one element at a time. Before you can pop from a heap, however, you need a valid heap to work with in the first place. The heapify function transforms any ordinary list into a properly structured min-heap in a single efficient pass. This is the recommended starting point for bulk heap construction, and the next section covers its usage in detail.

heapify()

Transforms a list into a valid min-heap in-place. This operates in O(n) time, which is more efficient than pushing items one by one.

Signature

heapq.heapify(x)

Rather than pushing items one at a time with heappush, which costs O(n log n) to build a heap from scratch, heapify rearranges an existing list in O(n) time. This linear-time construction is a notable algorithmic property of binary heaps and is the reason heapify should always be preferred when you already have all the data available upfront. The examples below illustrate basic heapification and compare performance against the repeated-push alternative.

Parameters

Parameter	Type	Default	Description
x	list	—	The list to transform into a heap (modified in-place)

Examples

Converting a list to a heap

import heapq

# Random order list
data = [9, 1, 7, 3, 5, 2, 8, 4, 6]

# Convert to heap in-place
heapq.heapify(data)

print(data)
# [1, 3, 2, 4, 5, 7, 8, 9, 6]

# Verify: smallest at index 0
print(data[0])
# 1

The heapify call in the previous example took a randomly ordered list and rearranged it so the smallest element sits at index zero and every parent node is smaller than its children. This single-call approach to heap construction is both cleaner and faster than looping over a collection with heappush. The following benchmark confirms the performance difference with concrete timing numbers on a thousand-element list.

Efficient alternative to repeated heappush

import heapq
import time

# Method 1: Push one by one
data1 = list(range(1000, 0, -1))
start = time.time()
heap1 = []
for item in data1:
    heapq.heappush(heap1, item)
time1 = time.time() - start

# Method 2: heapify
data2 = list(range(1000, 0, -1))
start = time.time()
heapq.heapify(data2)
time2 = time.time() - start

print(f"Heappush loop: {time1:.4f}s")
print(f"Heapify: {time2:.4f}s")
# heapify is significantly faster for bulk operations

The timing comparison above shows why heapify is the right tool for bulk heap construction. Once you have a working heap, certain workloads call for a combined push-then-pop operation rather than two separate calls. The heappushpop function performs both steps in a single atomic operation that runs more efficiently than calling heappush followed by heappop individually. The next section explains how this function works and when to reach for it.

heappushpop()

Pushes an item onto the heap and then pops the smallest item. This combined operation is more efficient than calling heappush() and heappop() separately.

Signature

heapq.heappushpop(heap, item)

The heappushpop routine pushes a new item onto the heap and then immediately pops and returns the smallest item from the resulting collection. If the newly pushed item is smaller than everything already in the heap it will be the one popped and returned. If the pushed item is larger than the current minimum then the old minimum is returned instead and the new item stays in the heap. The examples below show both scenarios.

Parameters

Parameter	Type	Default	Description
heap	list	—	The heap to operate on
item	any	—	The item to push before popping

Examples

More efficient than separate push and pop

import heapq

heap = [1, 2, 3, 4, 5]
heapq.heapify(heap)

# Push 0, then pop smallest (which is 0)
result = heapq.heappushpop(heap, 0)
print(f"Returned: {result}")
# Returned: 0
print(heap)
# [1, 2, 3, 4, 5]

# If item is larger than smallest, returns the smaller
heap2 = [1, 2, 3]
result2 = heapq.heappushpop(heap2, 10)
print(f"Returned: {result2}")
# Returned: 1
print(heap2)
# [2, 3, 10]

The previous examples demonstrated the fundamental behaviors of heappushpop with simple integer values. A practical scenario where this combined operation really shines is in streaming data processing, where you need to track only the k largest or smallest items seen so far from an unbounded stream without storing every value in memory. The next snippet implements this pattern concisely.

Use case: maintaining top-k elements

import heapq

def keep_top_k(stream, k):
    """Keep only the k largest items from a stream"""
    heap = []
    for item in stream:
        heapq.heappushpop(heap, item)
    return sorted(heap, reverse=True)

data = [5, 1, 3, 9, 2, 8, 4, 7, 6]
print(keep_top_k(data, 3))
# [9, 8, 7]

The top-k pattern above uses the heap to efficiently discard values that fall outside the desired window without ever sorting the full dataset. A closely related operation is heapreplace, which solves a slightly different problem by popping the current minimum first and then pushing a new item, keeping the total heap size constant. This distinction matters when you know the heap is nonempty and you want the smallest existing item returned before the new item is added.

heapreplace()

Pops the smallest item from the heap and pushes a new item. The heap size remains the same. Unlike heappushpop(), this pops first, then pushes.

Signature

heapq.heapreplace(heap, item)

Parameters

Parameter	Type	Default	Description
heap	list	—	The heap to operate on (must not be empty)
item	any	—	The item to push after popping

Examples

Fixed-size sliding window

import heapq

# Sliding window of size 3 - keep track of 3 smallest
window = [9, 5, 7]
heapq.heapify(window)

new_values = [3, 8, 1, 4, 6]

for val in new_values:
    popped = heapq.heapreplace(window, val)
    print(f"Replaced {popped} with {val}, window now: {window}")

# Output:
# Replaced 9 with 3, window now: [3, 5, 7]
# Replaced 3 with 8, window now: [5, 7, 8]
# Replaced 5 with 1, window now: [1, 7, 8]
# Replaced 1 with 4, window now: [4, 7, 8]
# Replaced 4 with 6, window now: [6, 7, 8]

The sliding window example above shows heapreplace maintaining a fixed-size min-heap that tracks the three smallest values seen so far, replacing the current smallest whenever a smaller value arrives. The key behavioral difference between heapreplace and heappushpop is the order of operations, and this ordering affects the return value when the pushed item happens to be smaller than every existing element. The next comparison makes this distinction explicit.

Difference from heappushpop

import heapq

heap = [1, 2, 3]

# heappushpop: pushes first, then pops smallest of all
result1 = heapq.heappushpop(heap, 0)
print(f"heappushpop({0}): {result1}, heap: {heap}")
# heappushpop(0): 0, heap: [1, 2, 3]

# heapreplace: pops first, then pushes
heap2 = [1, 2, 3]
result2 = heapq.heapreplace(heap2, 0)
print(f"heapreplace({0}): {result2}, heap: {heap2}")
# heapreplace(0): 1, heap: [0, 2, 3]

The side-by-side comparison above clarifies that heapreplace always pops before pushing, whereas heappushpop pushes first and then pops. With these fundamental heap mutation operations covered, the module also provides higher-level convenience functions that build and consume heaps internally so you do not have to manage them yourself. The nlargest and nsmallest functions, covered next, answer the common need to extract top or bottom k elements from arbitrary iterables.

nlargest()

Returns a list with the n largest elements from the iterable. Uses a heap internally for efficiency.

Signature

heapq.nlargest(n, iterable, key=None)

Parameters

Parameter	Type	Default	Description
n	int	—	Number of largest elements to return
iterable	iterable	—	The iterable to search
key	callable	None	Function to extract comparison key from each element

Examples

Finding top elements

import heapq

data = [1, 5, 2, 8, 3, 9, 4, 7, 6]

# Get 3 largest
print(heapq.nlargest(3, data))
# [9, 8, 7]

# Get top 5
print(heapq.nlargest(5, data))
# [9, 8, 7, 6, 5]

The examples above retrieve the largest integers from a simple numeric list. In real programs, the data is rarely a flat list of numbers. More often you have dictionaries, objects, or tuples and you need to rank them by a specific field or computed property. The key parameter accepts any callable that extracts the comparison value from each element, making nlargest and nsmallest flexible enough for structured data.

Using key function

import heapq

# List of dictionaries
products = [
    {'name': 'laptop', 'price': 999},
    {'name': 'phone', 'price': 699},
    {'name': 'tablet', 'price': 449},
    {'name': 'watch', 'price': 299},
    {'name': 'earbuds', 'price': 149},
]

# Get 2 most expensive
top_expensive = heapq.nlargest(2, products, key=lambda p: p['price'])
print(top_expensive)
# [{'name': 'laptop', 'price': 999}, {'name': 'phone', 'price': 699}]

The key function example above uses a lambda to extract the price field from product dictionaries so that nlargest can rank items by cost rather than by dictionary identity. When the problem calls for the opposite direction, nsmallest provides the same interface for retrieving the smallest n elements. The next section covers nsmallest with examples that mirror the nlargest patterns so you can see how both directions work on the same kinds of data.

nsmallest()

Returns a list with the n smallest elements from the iterable. Uses a heap internally for efficiency.

Signature

heapq.nsmallest(n, iterable, key=None)

Parameters

Parameter	Type	Default	Description
n	int	—	Number of smallest elements to return
iterable	iterable	—	The iterable to search
key	callable	None	Function to extract comparison key from each element

Examples

Finding smallest elements

import heapq

data = [9, 1, 4, 7, 3, 2, 8, 5, 6]

# Get 3 smallest
print(heapq.nsmallest(3, data))
# [1, 2, 3]

# Get 5 smallest
print(heapq.nsmallest(5, data))
# [1, 2, 3, 4, 5]

The basic nsmallest calls above return the smallest integers from a flat list, which is the simplest possible usage. When you work with nested data structures, the key parameter becomes essential for specifying which field drives the ordering. The example below applies nsmallest to a list of student dictionaries ranked by grade so you can identify the students who need the most attention.

Using key with complex data

import heapq

# Students with grades
students = [
    {'name': 'Alice', 'grade': 85},
    {'name': 'Bob', 'grade': 92},
    {'name': 'Carol', 'grade': 78},
    {'name': 'Dave', 'grade': 88},
    {'name': 'Eve', 'grade': 95},
]

# Get bottom 2 by grade
bottom_students = heapq.nsmallest(2, students, key=lambda s: s['grade'])
print(bottom_students)
# [{'name': 'Carol', 'grade': 78}, {'name': 'Alice', 'grade': 85}]

With the individual heapq functions covered, the module documentation now turns to real-world patterns that combine multiple heapq primitives to solve practical problems. These patterns go beyond single-function examples and show how the building blocks fit together to tackle tasks like priority scheduling with tie-breaking, merging sorted data streams from multiple sources, and computing running statistics over a series of values.

Common Patterns

Priority queue with task ordering

When multiple tasks have the same priority, use a counter to maintain insertion order:

import heapq
import itertools

class PriorityQueue:
    def __init__(self):
        self._heap = []
        self._counter = itertools.count()
    
    def push(self, priority, task):
        count = next(self._counter)
        heapq.heappush(self._heap, (priority, count, task))
    
    def pop(self):
        if not self._heap:
            raise IndexError("pop from empty queue")
        priority, count, task = heapq.heappop(self._heap)
        return priority, task

pq = PriorityQueue()
pq.push(2, 'task A')
pq.push(1, 'task B')
pq.push(1, 'task C')
pq.push(3, 'task D')

print(pq.pop())  # (1, 'task B')
print(pq.pop())  # (1, 'task C')
print(pq.pop())  # (2, 'task A')

The priority queue class above wraps heapq operations behind a clean interface that handles same-priority tasks correctly by using a monotonically increasing counter as a tiebreaker. A different kind of problem arises when you have multiple independently sorted sequences and you want to produce a single combined sorted output without loading everything into memory and running a full sort. The heapq.merge function solves exactly this scenario.

Merging sorted streams

The merge function efficiently combines multiple sorted iterables:

import heapq

# Multiple sorted log files
log1 = [1, 4, 7, 10]
log2 = [2, 5, 8, 11]
log3 = [3, 6, 9, 12]

# Merge in sorted order
merged = list(heapq.merge(log1, log2, log3))
print(merged)
# [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]

# Works with different sources - sorted timestamps from multiple sources
import itertools

timestamps = [
    (2024, 1, 15), (2024, 1, 20), (2024, 1, 25),
    (2024, 1, 18), (2024, 1, 22),
]
# Simulate sorted iterables
sorted_streams = [iter(ts) for ts in timestamps]
for item in heapq.merge(*sorted_streams):
    print(item)

The merge example above shows how to combine sorted timestamp streams from separate log sources into one ordered sequence. The final common pattern is the running median, which requires a more creative use of two heaps working together. A max-heap tracks the lower half of values and a min-heap tracks the upper half, with the median computed from the roots of both heaps. This technique updates in O(log n) per new value rather than requiring a full re-sort.

Running median with two heaps

import heapq

def running_median(stream):
    """Track running median using two heaps"""
    low = []   # max-heap (negative values)
    high = []  # min-heap
    
    for x in stream:
        # Push to low heap (negate for max-heap behavior)
        heapq.heappush(low, -heapq.heappushpop(high, x))
        
        # Balance heaps
        if len(low) > len(high):
            heapq.heappush(high, -heapq.heappop(low))
        
        # Calculate median
        if len(high) > len(low):
            median = float(high[0])
        else:
            median = (-low[0] + high[0]) / 2
        
        yield median

data = [5, 15, 1, 3, 8, 7, 9, 2]
print(list(running_median(data)))
# [5.0, 10.0, 5.0, 4.0, 5.0, 6.0, 7.0, 5.0]

See Also

• collections
• dataclasses
• itertools