Python PriorityQueue: A Practical Guide

The examples below handle equal priorities, custom objects, updates, deletion, max-heaps, and clean worker shutdown.

A Python priority queue normally returns the lowest-valued entry first; choose heapq for local algorithms, queue.PriorityQueue for threads, or asyncio.PriorityQueue for async tasks.

Python Priority Queues in 60 Seconds

A regular queue uses first in, first out, usually shortened to FIFO. The first item added is the first item removed. A stack uses last in, first out, or LIFO. The most recently added item is removed first.

A priority queue follows a different rule: remove the item with the most urgent stored priority. In Python’s standard min-priority queues, a smaller value comes out before a larger one. If priority 1 means urgent and priority 5 means routine, the entry with priority 1 is retrieved first even if it was added later.

The vocabulary can be confusing, so this guide uses these definitions consistently:

• Min-priority queue: the smallest stored priority value comes out first.
• Max-priority queue: the largest stored priority value comes out first.
• Higher urgency: a smaller number, unless a section explicitly uses a max-heap.
• Stable tie handling: entries with the same numeric priority come out in insertion order.

A heap is the usual data structure behind a priority queue. It keeps the next item at the root without fully sorting every entry. Think of it as a partially arranged waiting room: the next patient is always known, but the remaining patients are arranged only enough to find later patients efficiently.

Priority queues appear in job schedulers, event processing, shortest-path algorithms, and graph search. They also connect directly to the broader choices covered in the essential data structures guide.

Choose the Right Python Priority Queue

Python has three standard choices that look similar at first but serve different execution models.

Choice	Use it for	Coordination	Lowest value first
heapq	Algorithms and local data structures	You manage access	Yes
queue.PriorityQueue	Producer and consumer threads	Thread synchronization is included	Yes
asyncio.PriorityQueue	Tasks using async and await	Async task coordination	Yes

Use heapq when one flow of code owns the heap. Common examples include pathfinding, simulations, and a scheduler used from a single thread. heapq operates on a normal Python list. It does not add locking around that list.

Use queue.PriorityQueue when multiple threads put and get work through the same queue. The queue module supplies the locking behavior required for multi-producer, multi-consumer threaded programs. A positive maxsize also lets producers wait when the queue is full.

Use asyncio.PriorityQueue when the producers and consumers are asynchronous tasks. Its put() and get() operations can be awaited, but it is not intended to make access from multiple operating-system threads safe. Async queue methods do not accept a timeout argument; wrap the operation with asyncio.wait_for() if it needs a deadline.

An unsynchronized list-backed heap is often all an algorithm needs. Adding a synchronized queue where no threads share it introduces behavior such as blocking and task tracking that the algorithm may not need. The broader Python priority queue guide explores threaded and async coordination in more depth.

Build a Min-Priority Queue With heapq

The heapq module treats a list as a min-heap. Its invariant says that every parent is less than or equal to its children. As a result, heap[0] is always the smallest entry.

This complete example starts with an ordinary list, rearranges it with heapify(), pushes another job, peeks at the next job, and then removes every job:

import heapq

jobs = [
    (4, "write documentation"),
    (1, "repair production"),
    (3, "run tests"),
]

heapq.heapify(jobs)
heapq.heappush(jobs, (2, "review patch"))

print("next:", jobs[0])

while jobs:
    priority, task = heapq.heappop(jobs)
    print(priority, task)

next: (1, 'repair production')
1 repair production
2 review patch
3 run tests
4 write documentation

heapify() transforms the list in place in linear time. It is the right starting operation when all initial entries already exist. Repeatedly calling heappush() is useful when entries arrive over time.

The essential operations are:

Operation	Meaning	Cost
heapq.heapify(items)	Turn a list into a heap	O(n)
heapq.heappush(heap, item)	Add one item	O(log n)
heapq.heappop(heap)	Remove and return the smallest item	O(log n)
heap[0]	Peek without removing	O(1)

Calling heappop() on an empty heap raises IndexError. Reading heap[0] on an empty heap also fails because the list has no first element, so check the list before either operation when emptiness is possible.

The tuple entries work because Python compares tuples from left to right. It compares the numeric priorities first. Only when those values are equal does it examine the second fields.

That second comparison is convenient for strings, but it does not provide FIFO tie handling:

import heapq

tasks = []
heapq.heappush(tasks, (1, "second"))
heapq.heappush(tasks, (1, "first"))

print(heapq.heappop(tasks))
print(heapq.heappop(tasks))

(1, 'first')
(1, 'second')

Although "second" was inserted first, "first" comes out first because string comparison breaks the tie. The next section fixes both unstable ties and payloads that cannot be compared.

Handle Max Priorities, Ties, and Custom Objects

Python 3.14 adds five native max-heap functions:

• heapify_max() creates a max-heap in place.
• heappush_max() adds an entry.
• heappop_max() removes the largest entry.
• heapreplace_max() removes the largest entry and then adds a replacement.
• heappushpop_max() adds an entry and then removes the largest entry.

This Python 3.14 example uses the largest numeric value as the next priority:

import heapq

priorities = [4, 1, 3, 2]
heapq.heapify_max(priorities)
heapq.heappush_max(priorities, 7)

print(heapq.heappop_max(priorities))
print(priorities[0])

7
4

Label max-heap code clearly because these names are unavailable on Python 3.13 and earlier. For a project that supports those versions, negate each numeric priority and use the regular min-heap API:

import heapq
from itertools import count

sequence = count()
jobs = []

def push_max(priority, task):
    heapq.heappush(jobs, (-priority, next(sequence), task))

def pop_max():
    negative_priority, _, task = heapq.heappop(jobs)
    return -negative_priority, task

push_max(4, "normal")
push_max(9, "urgent")
push_max(9, "urgent follow-up")

print(pop_max())
print(pop_max())

(9, 'urgent')
(9, 'urgent follow-up')

Here a larger original number means greater urgency. Negation turns 9 into -9, which the min-heap removes before -4. The sequence number preserves insertion order between the two entries whose original priority is 9.

For a min-heap, including a max-priority queue implemented with negated priorities, that (priority, sequence, item) shape is the safest general entry format. The sequence comes from itertools.count(), so every entry receives a unique increasing number. Python can settle every tie by comparing those integers and never needs to compare the payloads. With Python 3.14’s native max-heap functions, use (priority, -next(sequence), item) instead, because the larger sequence value would otherwise be removed first.

Without the sequence field, two custom objects with equal priorities may fail during comparison:

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

try:
    print((1, Task("first")) < (1, Task("second")))
except TypeError:
    print("Task objects cannot break the tie")

Task objects cannot break the tie

The stable three-field entry works with those same objects:

import heapq
from itertools import count

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

sequence = count()
heap = []

heapq.heappush(heap, (1, next(sequence), Task("first")))
heapq.heappush(heap, (1, next(sequence), Task("second")))

print(heapq.heappop(heap)[2].name)
print(heapq.heappop(heap)[2].name)

first
second

A dataclass provides another readable representation when entries need named fields:

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    sequence: int
    item: Any = field(compare=False)

order=True makes the priority and sequence comparable. compare=False excludes item, so arbitrary payload objects do not participate in ordering. The sequence remains necessary when equal priorities must be FIFO.

Update or Remove Priorities Safely

A heap supports efficient changes at its root, but it does not provide a direct operation for finding and rewriting any arbitrary entry. Searching for an entry and repairing its position manually would complicate the implementation.

The standard pattern combines four pieces:

• An entry-finder dictionary maps each active task to its current heap entry.
• A unique counter supplies stable tie-breaking numbers.
• A private marker identifies removed entries.
• Pop operations discard marked entries until they find an active one.

An update is therefore two operations: invalidate the old entry, then push a replacement. This reusable scheduler uses smaller numbers for greater urgency:

import heapq
from itertools import count

REMOVED = object()

class PriorityScheduler:
    def __init__(self):
        self._heap = []
        self._entries = {}
        self._sequence = count()

    def add(self, task, priority):
        if task in self._entries:
            self.remove(task)

        entry = [priority, next(self._sequence), task]
        self._entries[task] = entry
        heapq.heappush(self._heap, entry)

    def update(self, task, priority):
        self.add(task, priority)

    def remove(self, task):
        entry = self._entries.pop(task)
        entry[2] = REMOVED

    def pop(self):
        while self._heap:
            priority, _, task = heapq.heappop(self._heap)

            if task is REMOVED:
                continue

            del self._entries[task]
            return priority, task

        raise KeyError("pop from an empty priority scheduler")

    def __len__(self):
        return len(self._entries)


scheduler = PriorityScheduler()
scheduler.add("send email", 5)
scheduler.add("run linter", 2)
scheduler.update("send email", 1)
scheduler.add("backup files", 3)
scheduler.remove("backup files")

print(scheduler.pop())
print(scheduler.pop())
print(len(scheduler))

(1, 'send email')
(2, 'run linter')
0

The first "send email" entry remains inside _heap after the update, but its task field contains REMOVED. The replacement entry with priority 1 is active in _entries. When the stale entry eventually reaches the root, pop() skips it.

This is called lazy deletion. The underlying list can temporarily contain more entries than len(scheduler) reports because removed and replaced entries remain until popped. The dictionary represents active work; the heap may also hold stale bookkeeping entries.

Tasks used as dictionary keys must be hashable. Strings, integers, and immutable identifiers work naturally. For a mutable task object, use a stable task ID as the dictionary key and keep the full object as payload data.

This scheduler is suitable when one execution flow owns it. If several threads must call these methods concurrently, use synchronization or place the work behind queue.PriorityQueue.

Use PriorityQueue With Worker Threads

queue.PriorityQueue supplies a synchronized priority queue for producer and consumer threads. Its entries still follow lowest-value-first ordering, so the stable (priority, sequence, task) format remains useful.

A positive maxsize makes put() block while the queue is full. The producer resumes after a worker removes an entry. A value of zero or less creates an unbounded queue.

The following complete example targets Python 3.13 or later because it uses Queue.shutdown() and catches queue.ShutDown:

import queue
import threading
from itertools import count

work = queue.PriorityQueue(maxsize=2)
sequence = count()

def process(task):
    print(f"{threading.current_thread().name}: {task}")

def worker():
    while True:
        try:
            priority, _, task = work.get()
        except queue.ShutDown:
            return

        try:
            process(task)
        except Exception as error:
            print(f"failed: {task}: {error}")
        finally:
            work.task_done()

threads = [
    threading.Thread(target=worker, name=f"worker-{number}")
    for number in range(1, 3)
]

for thread in threads:
    thread.start()

for priority, task in [
    (3, "generate report"),
    (1, "repair service"),
    (2, "run tests"),
    (2, "publish package"),
]:
    work.put((priority, next(sequence), task))

work.shutdown()
work.join()

for thread in threads:
    thread.join()

There is no output block because thread scheduling can change which worker prints each line and when a producer’s later item becomes available. A priority queue selects the best item available when get() runs; it does not wait for a future higher-priority item or preempt work that a worker has already retrieved. The sequence field fixes retrieval order for equal priorities, but concurrent processing can still make their printed output appear in either order. The code does guarantee that each retrieved item receives one matching task_done() call, even if process() raises an exception.

During normal operation and graceful shutdown, join() waits until every enqueued item has been retrieved and marked complete. Putting task_done() in finally prevents a processing failure from leaving the unfinished-task count stuck. An immediate shutdown with shutdown(immediate=True) can instead unblock join() without every item being processed.

shutdown() tells blocked or future queue operations that the queue is closing. Once the available work has been consumed, get() raises queue.ShutDown, and each worker exits. Both Queue.shutdown() and queue.ShutDown were added in Python 3.13. Projects supporting older versions need an explicitly designed sentinel protocol instead.

Do not write while not work.empty() as the worker condition. empty(), full(), and qsize() report an observed state only. Another thread can change the queue immediately afterward, so those checks cannot promise that the next get() or put() will avoid blocking.

For async tasks, the selection rule stays the same: use asyncio.PriorityQueue, then await put() and get(). This runnable example requires Python 3.13 or later:

import asyncio
from itertools import count

sequence = count()

async def producer(work):
    for priority, task in [
        (3, "generate report"),
        (1, "repair service"),
        (2, "run tests"),
        (2, "publish package"),
    ]:
        await work.put((priority, next(sequence), task))
    work.shutdown()

async def worker(work, name):
    while True:
        try:
            _, _, task = await work.get()
        except asyncio.QueueShutDown:
            return

        try:
            print(f"{name}: {task}")
        finally:
            work.task_done()

async def main():
    work = asyncio.PriorityQueue()
    workers = [
        asyncio.create_task(worker(work, f"worker-{number}"))
        for number in range(1, 3)
    ]

    try:
        await producer(work)
        await work.join()
        await asyncio.gather(*workers)
    finally:
        for task in workers:
            if not task.done():
                task.cancel()
        await asyncio.gather(*workers, return_exceptions=True)

asyncio.run(main())

The producer calls shutdown() after submitting its last entry. Once the queue is empty, blocked or future get() calls raise asyncio.QueueShutDown, allowing the workers to exit. The finally block also cancels and collects any workers left behind if the main task fails. Both asyncio.Queue.shutdown() and asyncio.QueueShutDown were added in Python 3.13; older versions need an explicitly designed sentinel protocol. Use asyncio.wait_for(work.get(), timeout) when a get operation needs a timeout.

Thread synchronization and async coordination solve different problems. queue.PriorityQueue protects shared access between threads. asyncio.PriorityQueue lets async tasks pause cooperatively while they wait for queue capacity or work.

Complexity, Pitfalls, and Interview Checklist

For a list-backed heap, heapify() takes O(n), peeking at heap[0] takes O(1), and each push or pop takes O(log n). A priority queue does not keep the entire list sorted. It maintains enough order to expose and remove the root efficiently.

Before using or explaining one, check these points:

• State the priority direction. Standard Python min-heaps retrieve the lowest value first. A native or negated max-heap retrieves the largest original value first.
• Add a sequence number for stable ties. Use an increasing sequence for min-heaps and negated-priority max-heaps, but negate the sequence for a native max-heap.
• Prevent payload comparison. Use (priority, tie_breaker, item) or a dataclass field with compare=False.
• Handle empty heaps. heappop() raises IndexError when no entry exists.
• Treat queue state checks as snapshots. Do not base threaded correctness on empty(), full(), or qsize().
• Match every threaded get() with task_done(). Put it in finally, then use join() to wait for completion.
• Expect stale entries after lazy deletion. They remain in the heap until a later pop skips them.
• Choose synchronization only when required. Use heapq for locally owned algorithmic state, queue.PriorityQueue for threads, and asyncio.PriorityQueue for async tasks.

These details appear in Python interviews because they test more than API recall. They reveal whether the implementation handles ordering, object comparison, mutation, and concurrency correctly. The Python interview guide places priority queues alongside the other language and data-structure topics worth practising.

Where the books fit

For a quick API refresher, Python in One Week covers the core collection and queue operations. Python in Three Months fits readers building algorithm and interview depth, including heaps, graph search, and concurrency patterns. For design work involving schedulers, backpressure, threaded services, and async coordination, Python for Staff Engineers carries the same choices into system-level trade-offs. Pick one execution model, implement the stable entry format, and test the failure paths before adding more machinery.