Priority Queue in Python: heapq, Threads, and Async

A Python priority queue should use heapq for an ordinary in-memory data structure, queue.PriorityQueue for threads, or asyncio.PriorityQueue for async tasks. Each option retrieves the smallest entry first, but only the queue classes provide worker coordination such as waiting for items and tracking completed work.

Use heapq for locally owned data, queue.PriorityQueue for threaded workers, and asyncio.PriorityQueue for async workers, with a sequence number whenever equal priorities or non-comparable payloads are possible.

Which Python Priority Queue Should You Use?

A priority queue retrieves items according to a priority value instead of strictly following insertion order. Python has three standard choices because storing prioritized data and coordinating concurrent workers are different jobs.

Situation	Use	What it provides
One function or object owns the data	heapq	Operations on a regular Python list
Multiple threads produce or consume work	queue.PriorityQueue	Waiting, locking, capacity limits, and completion tracking
Multiple async tasks produce or consume work	asyncio.PriorityQueue	Awaitable queue operations and completion tracking

Use heapq when you need the data structure itself. It is a module of functions that maintain a heap inside a list. It does not provide worker synchronization, timeouts, or task completion accounting.

Use queue.PriorityQueue when operating-system threads share pending work. Its put() and get() operations can block, meaning the calling thread waits until it can proceed. A bounded queue also provides backpressure: producers wait when the queue has reached its configured capacity, preventing them from creating work faster than consumers can accept it.

Use asyncio.PriorityQueue when coroutines share pending work through async and await. It belongs in async code and is not thread-safe. Its methods do not accept timeout arguments, so timeouts are applied with asyncio.wait_for().

A third-party indexed heap can be appropriate when a program must update known keys directly and frequently. That is a specialized requirement. For most schedulers, the standard library’s documented lazy-deletion pattern is simpler and sufficient.

The choice is therefore about ownership and execution model, not about three interchangeable spellings. If the surrounding program is still taking shape, the Python learning roadmap helps place data structures before threads and async coordination.

How a Python Priority Queue Works

A first-in, first-out queue, usually called a FIFO queue, returns the oldest item. A priority queue returns the item with the smallest priority value under Python’s default convention. Given priorities 1, 4, and 9, Python retrieves 1 first.

The standard implementations use heap ordering. A min-heap is a tree-shaped data structure in which every parent is no greater than its children. The first element, called the root, is therefore the smallest element. In a heapq list, the root is stored at index 0.

The heap invariant is the ordering rule that keeps the smallest item at the root. It does not require every element to be globally sorted. For example, a valid heap list might display as [1, 3, 2, 8, 7, 6]. The root is guaranteed to be smallest, but the remaining positions should not be read as sorted output.

A heap and a priority queue are closely related but are not identical concepts:

• The priority queue defines the behavior: insert an item and retrieve the next item by priority.
• The heap is the data structure used to implement that behavior.
• heapq exposes heap operations on a list.
• queue.PriorityQueue and asyncio.PriorityQueue add coordination around prioritized entries.

A heap is not stable by itself. Stable ordering means that two items with equal priorities leave in the order they entered. If entries contain only (priority, task), Python compares task when two priorities tie. That can reorder comparable values or raise TypeError for payload objects that cannot be ordered.

The reliable entry shape is:

(priority, sequence_number, task)

The priority is compared first. The unique sequence number breaks ties, so Python never needs to compare the task.

Build a Priority Queue With heapq

The following standalone example introduces the heap operations used by a small task scheduler. These tasks begin as an ordinary list, so heapify() establishes the heap invariant in place.

import heapq

tasks = [
    (5, "archive reports"),
    (1, "restore service"),
    (3, "send invoices"),
]

heapq.heapify(tasks)

print(tasks[0])
heapq.heappush(tasks, (2, "answer support ticket"))

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

(1, 'restore service')
(1, 'restore service')
(2, 'answer support ticket')
(3, 'send invoices')
(5, 'archive reports')

heapq.heapify(tasks) transforms the existing list into a heap in place. tasks[0] peeks at the smallest entry without removing it. heappush() inserts while preserving the heap invariant, and heappop() removes and returns the smallest entry.

Calling heappop() on an empty heap raises IndexError. Check whether the list contains an item or catch that exception when an empty heap is a normal outcome.

Two combined operations are useful in fixed-size or streaming algorithms:

import heapq

heap = [(2, "normal"), (5, "later")]
heapq.heapify(heap)

returned = heapq.heappushpop(heap, (1, "urgent"))
print(returned)
print(heap[0])

replaced = heapq.heapreplace(heap, (4, "new"))
print(replaced)
print(heap[0])

(1, 'urgent')
(2, 'normal')
(2, 'normal')
(4, 'new')

heappushpop() pushes an entry and then returns the smallest entry. In this example, the new priority 1 entry comes straight back, leaving the earlier heap intact.

heapreplace() removes the current smallest entry first and then adds the replacement. It requires a non-empty heap. The difference matters when the incoming item is smaller than the current root.

Do not sort the heap after each insertion. Sorting would produce a sorted list, but the next heap operation needs only the heap invariant. To retrieve every item in priority order, repeatedly call heappop().

The scheduler above uses strings, so tied tuples can still compare their second fields. Real task payloads are often instances containing callbacks, request data, or other objects. The next entry format removes that hidden dependency. For a broader comparison of container behavior, see lists, tuples, sets, and dictionaries in Python.

Handle Equal Priorities and Custom Objects

Consider two tasks whose payload objects do not define ordering:

from dataclasses import dataclass

@dataclass
class Task:
    name: str

Entries such as (2, Task("index")) and (2, Task("email")) tie on priority. Tuple comparison then reaches the two Task instances, and the heap operation can raise TypeError.

Add a unique sequence number between the priority and payload:

import heapq
from dataclasses import dataclass
from itertools import count

@dataclass
class Task:
    name: str

sequence = count()
heap = []

heapq.heappush(heap, (2, next(sequence), Task("index")))
heapq.heappush(heap, (1, next(sequence), Task("repair")))
heapq.heappush(heap, (2, next(sequence), Task("email")))

while heap:
    priority, _, task = heapq.heappop(heap)
    print(priority, task.name)

1 repair
2 index
2 email

itertools.count() returns a new increasing integer for each call to next(). No two live entries receive the same sequence number. Equal priorities therefore use insertion order, and tuple comparison stops before reaching the payload.

A dataclass can express the same comparison rule when named fields are preferable:

from dataclasses import dataclass, field
from typing import Any

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

The generated ordering compares priority and then sequence. compare=False excludes task, which allows arbitrary payload objects.

The three-field tuple is compact and works well inside a reusable queue class. The dataclass form makes a public entry type easier to read. Both follow the same rule: the payload must never decide the order.

Update, Cancel, and Reverse Priorities Safely

Changing an entry’s priority in place can break the heap invariant. Removing an arbitrary entry also leaves a hole that must be repaired. The standard lazy-deletion pattern avoids both problems:

1. Store the current heap entry for each task key in an entry-finder dictionary.
2. To update a task, mark its old entry as removed and push a replacement.
3. To cancel a task, mark its current entry as removed.
4. While popping, discard removed or stale entries until a live task appears.

Here is the canonical scheduler implementation used by the tests later in this guide:

import heapq
from itertools import count
from typing import Any, Hashable


class EmptyPriorityQueue(Exception):
    pass


class StablePriorityQueue:
    def __init__(self) -> None:
        self._heap: list[list[Any]] = []
        self._entries: dict[Hashable, list[Any]] = {}
        self._sequence = count()
        self._removed = object()

    def __len__(self) -> int:
        return len(self._entries)

    def add(
        self,
        key: Hashable,
        task: Any,
        priority: int | float,
    ) -> None:
        if key in self._entries:
            self.cancel(key)

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

    def cancel(self, key: Hashable) -> None:
        entry = self._entries.pop(key)
        entry[3] = self._removed

    def reprioritize(
        self,
        key: Hashable,
        new_priority: int | float,
    ) -> None:
        old_entry = self._entries[key]
        task = old_entry[3]
        self.cancel(key)
        self.add(key, task, new_priority)

    def peek(self) -> tuple[int | float, Hashable, Any]:
        self._discard_stale_roots()
        if not self._heap:
            raise EmptyPriorityQueue("priority queue is empty")

        priority, _, key, task = self._heap[0]
        return priority, key, task

    def pop(self) -> tuple[int | float, Hashable, Any]:
        while self._heap:
            priority, _, key, task = heapq.heappop(self._heap)

            if task is self._removed:
                continue
            if self._entries.get(key) is not None:
                del self._entries[key]
                return priority, key, task

        raise EmptyPriorityQueue("priority queue is empty")

    def _discard_stale_roots(self) -> None:
        while self._heap and self._heap[0][3] is self._removed:
            heapq.heappop(self._heap)

The unique object() sentinel cannot equal a real task supplied from outside the class. Identity testing with is distinguishes it from every payload, including strings that happen to contain words such as "removed".

Each task has a separate hashable key. This lets the payload itself remain unhashable or mutable. Calling add() with an existing key performs a replacement, while reprioritize() keeps the existing payload and changes only its numeric priority.

Canceled and replaced entries can remain inside _heap temporarily. They are harmless because pop() and peek() skip marked roots. The entry-finder contains only live tasks, so len(queue) reports active work rather than the heap list’s physical size.

For max-priority behavior, Python 3.14 adds dedicated operations including heapify_max(), heappush_max(), and heappop_max(). Those operations retrieve the largest entry first.

On older Python versions, negating numeric priorities is a straightforward fallback when the priority type supports negation:

import heapq

max_heap = []
heapq.heappush(max_heap, (-10, "large priority"))
heapq.heappush(max_heap, (-3, "small priority"))

negative_priority, task = heapq.heappop(max_heap)
print(-negative_priority, task)

10 large priority

Negation changes the stored representation, so convert the value back after popping. It is unsuitable for priority types that do not support unary negation.

Use Priority Queues With Threads or asyncio

The canonical StablePriorityQueue is appropriate when one part of the program controls access. It does not add locking. Threaded and async workers should use the queue class made for their execution model.

A threaded scheduler can preserve stable ties by storing (priority, sequence, task):

from itertools import count
from queue import Empty, PriorityQueue, ShutDown
from threading import Thread
from typing import Callable

Work = tuple[str, Callable[[], None]]

pending: PriorityQueue[tuple[int, int, Work]] = PriorityQueue(maxsize=100)
sequence = count()


def submit(priority: int, name: str, action: Callable[[], None]) -> None:
    pending.put((priority, next(sequence), (name, action)), timeout=1)


def worker() -> None:
    while True:
        try:
            _, _, (_, action) = pending.get(timeout=1)
        except Empty:
            continue
        except ShutDown:
            return

        try:
            action()
        finally:
            pending.task_done()


thread = Thread(target=worker)
thread.start()

submit(2, "send report", lambda: None)
submit(1, "restore service", lambda: None)

pending.join()
pending.shutdown()
thread.join()

This complete example requires Python 3.13 because it uses Queue.shutdown() and the ShutDown exception. A non-immediate shutdown prevents further growth and allows queued work to drain. A worker blocked in get() is released by shutdown and receives ShutDown once no more work can be retrieved.

put(..., timeout=1) waits for capacity in the bounded queue and raises Full if the timeout expires. get(timeout=1) waits for work and raises Empty if none arrives during the timeout.

Every successful get() must have one corresponding task_done(). The finally block preserves that pairing even if a task action raises an exception. join() waits until all items placed in the queue have been marked complete.

Do not use qsize(), empty(), or full() to decide that the next operation cannot block. Those methods report approximate state, which can change before the next put() or get().

The async version uses the same three-field entries, but its waiting operations are awaited:

import asyncio
from itertools import count
from typing import Awaitable, Callable

AsyncAction = Callable[[], Awaitable[None]]
STOP = object()


async def run_scheduler() -> None:
    pending: asyncio.PriorityQueue[tuple[int, int, object]] = (
        asyncio.PriorityQueue(maxsize=100)
    )
    sequence = count()

    async def worker() -> None:
        while True:
            try:
                _, _, action = await asyncio.wait_for(
                    pending.get(),
                    timeout=1,
                )
            except TimeoutError:
                continue

            try:
                if action is STOP:
                    return
                await action()
            finally:
                pending.task_done()

    async def restore_service() -> None:
        await asyncio.sleep(0)

    async def send_report() -> None:
        await asyncio.sleep(0)

    workers = [asyncio.create_task(worker()) for _ in range(2)]

    await pending.put((2, next(sequence), send_report))
    await pending.put((1, next(sequence), restore_service))
    await pending.join()

    for _ in workers:
        await pending.put((100, next(sequence), STOP))

    await pending.join()
    await asyncio.gather(*workers)


asyncio.run(run_scheduler())

asyncio.PriorityQueue has no timeout parameter on get() or put(). Wrapping an operation in asyncio.wait_for() adds a timeout. The queue is intended for async tasks and must not be used as a thread-safe bridge.

The unique STOP object provides graceful shutdown for this example. One sentinel is added for each worker, after ordinary work has completed. Each worker calls task_done() for its sentinel before returning, which allows the second join() to finish.

The threaded and async examples coordinate producers and consumers, but they do not implement direct reprioritization or cancellation of already queued entries. If those features are required concurrently, place the lazy-deletion policy behind a carefully synchronized scheduler boundary instead of reaching into either queue’s internal storage. The Python GIL guide explains why thread coordination is still necessary even when Python code execution has interpreter-level constraints.

Complexity, Common Bugs, and Tests

heapify() constructs a heap from an existing list in linear time. Peeking at heap[0] is constant time. Insertion and root removal follow the height of the heap, so they take logarithmic time. Lazy deletion keeps updates safe, although stale entries remain until they reach the root and are discarded.

The tests below exercise the canonical StablePriorityQueue implementation. They verify lowest-first retrieval, FIFO tie-breaking, custom payloads, reprioritization, cancellation, stale-entry skipping, and empty-queue behavior.

import unittest
from dataclasses import dataclass


@dataclass
class Task:
    name: str
    details: list[str]


class StablePriorityQueueTests(unittest.TestCase):
    def test_smaller_priority_is_popped_first(self) -> None:
        queue = StablePriorityQueue()
        queue.add("routine", Task("routine", []), 9)
        queue.add("urgent", Task("urgent", []), 1)

        self.assertEqual(queue.pop()[1], "urgent")
        self.assertEqual(queue.pop()[1], "routine")

    def test_equal_priorities_keep_fifo_order(self) -> None:
        queue = StablePriorityQueue()
        queue.add("first", Task("first", []), 2)
        queue.add("second", Task("second", []), 2)

        self.assertEqual(queue.pop()[1], "first")
        self.assertEqual(queue.pop()[1], "second")

    def test_non_comparable_payloads_do_not_get_compared(self) -> None:
        queue = StablePriorityQueue()
        queue.add("a", Task("a", ["mutable"]), 3)
        queue.add("b", Task("b", ["mutable"]), 3)

        self.assertEqual(queue.pop()[2].name, "a")
        self.assertEqual(queue.pop()[2].name, "b")

    def test_reprioritize_skips_stale_entry(self) -> None:
        queue = StablePriorityQueue()
        queue.add("report", Task("report", []), 8)
        queue.add("repair", Task("repair", []), 4)
        queue.reprioritize("report", 1)

        self.assertEqual(queue.pop()[:2], (1, "report"))
        self.assertEqual(queue.pop()[:2], (4, "repair"))

    def test_cancel_removes_live_task(self) -> None:
        queue = StablePriorityQueue()
        queue.add("keep", Task("keep", []), 2)
        queue.add("cancel", Task("cancel", []), 1)
        queue.cancel("cancel")

        self.assertEqual(queue.pop()[1], "keep")
        with self.assertRaises(EmptyPriorityQueue):
            queue.pop()

    def test_empty_peek_and_pop_raise_domain_error(self) -> None:
        queue = StablePriorityQueue()

        with self.assertRaises(EmptyPriorityQueue):
            queue.peek()
        with self.assertRaises(EmptyPriorityQueue):
            queue.pop()


if __name__ == "__main__":
    unittest.main()

Common priority-queue bugs include:

• Assuming the backing list is fully sorted instead of trusting only the root.
• Using (priority, task) when equal priorities and custom payload objects are possible.
• Editing an entry’s priority in place and breaking the heap invariant.
• Using a string as a removed marker, which a real payload could equal.
• Calling get() without guaranteeing a matching task_done().
• Treating empty() as proof that a later get() cannot block.
• Passing timeout= to an asyncio queue method instead of using asyncio.wait_for().
• Sharing asyncio.PriorityQueue between threads.

Priority queues often appear beside heaps and graph algorithms in Python interview preparation. The entry format and lazy-deletion tests matter more than memorizing method names because they expose the failure cases hidden by simple demonstrations.

Where the books fit

For a quick review of heap operations, Python in One Week is the compact reference tier. Python in Three Months fits readers building job-ready skill across data structures, testing, threads, and async code. For scheduler design, concurrency boundaries, and the tradeoffs behind production queue APIs, Python for Staff Engineers carries the topic into senior engineering decisions. Start with the execution model, keep ties stable, and test the paths where queued work changes or disappears.