Priority queue in Python: heapq, the size-k trick and when a problem needs one

Short answer. In Python, a priority queue is the heapq module over a plain list. heappush and heappop cost O(log n), reading the minimum costs O(1). It is a min-heap. For a max-heap you push negated values. Use it when a problem asks for "k largest", "k closest" or "merge k sorted lists".

What is a priority queue?

A queue gives you the oldest item. A priority queue gives you the item with the best priority. Best means smallest by default.

Under the hood it is a binary heap. That is a complete binary tree stored in an array. The root holds the minimum. Children of index i sit at 2i + 1 and 2i + 2.

Insert puts the item at the end and bubbles it up. Pop moves the last item to the root and sinks it down. Both walk one path of height log n.

How does the priority queue API in Python look?

Call	Cost	What it does
heapq.heapify(lst)	O(n)	Turns a list into a heap in place
heapq.heappush(h, x)	O(log n)	Adds x
heapq.heappop(h)	O(log n)	Removes and returns the smallest
h[0]	O(1)	Peeks at the smallest
heapq.heappushpop(h, x)	O(log n)	Push then pop, faster than two calls
heapq.nlargest(k, it)	O(n log k)	k largest items of any iterable

There is no decrease-key. To update a priority, push a new entry. Skip stale ones when they pop. This is called lazy deletion.

When does a problem need a priority queue?

• "k largest", "k smallest", "k most frequent", "k closest points".
• "Merge k sorted lists" or "smallest range covering k lists".
• "Median of a data stream" with two heaps.
• "Task scheduler", "meeting rooms", "next event by time".

The shared signal: items keep arriving, and you always need the best one. Sorting after every change costs O(n log n). The heap does it in O(log n).

How do you write a priority queue in Python?

import heapq
from collections import Counter

# max-heap through negation
h = []
for x in [5, 1, 8]:
    heapq.heappush(h, -x)
largest = -heapq.heappop(h)         # 8

# tuples: compared by the first field
tasks = []
heapq.heappush(tasks, (2, "write tests"))
heapq.heappush(tasks, (1, "fix build"))
priority, name = heapq.heappop(tasks)   # (1, "fix build")

def top_k_frequent(nums, k):
    count = Counter(nums)
    h = []
    for value, freq in count.items():
        heapq.heappush(h, (freq, value))
        if len(h) > k:
            heapq.heappop(h)            # drop the rarest
    return [value for freq, value in h]

The heap of size k is the standard trick. It keeps the k best seen so far. Memory O(k), time O(n log k).

What does one push look like?

Heap [1, 3, 2, 7, 5]. Push 0. It goes to the end: [1, 3, 2, 7, 5, 0]. Index 5, parent index 2, value 2. Zero is smaller, swap: [1, 3, 0, 7, 5, 2].

Now index 2, parent index 0, value 1. Zero is smaller, swap: [0, 3, 1, 7, 5, 2]. Two swaps, done. Pop reverses the moves. The last item goes to the root and sinks.

How does a priority queue solve Merge k Sorted Lists?

Push the head of every list as (value, list index, node). Pop the smallest and append it to the output. Push the next node of that list.

The heap never holds more than k nodes. Every node enters and leaves once. Total time O(N log k), where N counts all nodes.

The list index in the tuple breaks ties. Without it Python tries to compare two nodes and raises.

What mistakes break a priority queue solution?

• Treating the heap list as sorted. Only index 0 is guaranteed.
• Forgetting the minus sign and getting the k smallest.
• Pushing objects that cannot be compared. Wrap them in a tuple with a number first.
• Searching the heap linearly and calling it O(log n).

How do you explain a priority queue in an interview?

Say you need the current best while items keep changing. Give the costs: O(1) peek, O(log n) push and pop, O(n) build. Show the size-k trick.

Then name the alternative: sort every time at O(n log n). That comparison is the reason for the heap.

Drill the pattern against its neighbours on the binary heap page. The full map of patterns lives in LeetCode patterns.

FAQ

Is queue.PriorityQueue better than heapq?

It is thread safe and slower. For interviews and single threaded code, heapq is the right choice.

How do you remove an arbitrary item?

You cannot in O(log n) with heapq alone. Mark it as removed in a set. When it reaches the top, pop and ignore it.

How do you build a max-heap without negation?

Wrap items in a small class with a reversed __lt__. Or use heapq.nlargest when you need the k largest once.

Why is heapify O(n) and not O(n log n)?

Half the nodes are leaves and need no sinking. Higher levels have fewer nodes. The sum of all sinks is linear.