Of course! In Python, a Priority Queue is an abstract data type that operates similar to a regular queue, but with a key difference: each element has a "priority" associated with it. An element with higher priority is served before an element with lower priority.

If two elements have the same priority, they are typically served in the order they were added (this is called a FIFO queue for elements with the same priority).

Here's a complete guide to using Priority Queues in Python, from the standard library to third-party options.

The Standard Library: heapq Module

For most use cases, the heapq module is the best and most efficient way to implement a priority queue in Python. It provides an implementation of the min-heap algorithm.

Key Concept (Min-Heap): In a min-heap, the smallest element is always at the root. heapq makes this element easily accessible. To use it as a priority queue where the highest priority number is served first, you can simply store your priorities as negative numbers.

How to Use heapq

A heapq is a list that you maintain with special functions.

a) Initialization

You start with a standard list.

import heapq
# An empty priority queue
pq = []
# You can also initialize it with items
# Note: heapq doesn't guarantee the initial list is a heap, so you must heapify it.
initial_items = [ (3, 'task 3'), (1, 'task 1'), (2, 'task 2') ]
pq = initial_items.copy()
heapq.heapify(pq)
print(f"Initial heap: {pq}")
# Output might be: Initial heap: [(1, 'task 1'), (3, 'task 3'), (2, 'task 2')]
# The smallest element is at the front, but the rest are not fully sorted.

b) Adding Elements (heappush)

Use heapq.heappush() to add an element to the queue while maintaining the heap property. The element must be a tuple, with the first item being the priority.

heapq.heappush(pq, (1, 'task 1.1')) # Added another task with priority 1
heapq.heappush(pq, (4, 'task 4'))
heapq.heappush(pq, (0, 'urgent task'))
print(f"Queue after pushes: {pq}")
# Output: Queue after pushes: [(0, 'urgent task'), (1, 'task 1.1'), (2, 'task 2'), (4, 'task 4'), (3, 'task 3')]

c) Removing Elements (heappop)

Use heapq.heappop() to remove and return the element with the highest priority (the smallest value, which is the root of the heap).

# Get the highest priority task (lowest number)
priority, task = heapq.heappop(pq)
print(f"Popped: {priority} - {task}")
# Output: Popped: 0 - urgent task
print(f"Queue after pop: {pq}")
# Output: Queue after pop: [(1, 'task 1.1'), (3, 'task 3'), (2, 'task 2'), (4, 'task 4')]

If you pop again, you'll get the next highest priority item.

priority, task = heapq.heappop(pq)
print(f"Popped: {priority} - {task}")
# Output: Popped: 1 - task 1.1

d) Peeking at the Top Element (heap[0])

If you want to see the highest priority item without removing it, you can just look at the first element of the list.

print(f"Next highest priority task: {pq[0]}")
# Output: Next highest priority task: (2, 'task 2')

Full Example: Processing Tasks by Priority

Let's put it all together in a realistic example.

import heapq
# List of tasks: (priority, description)
# Lower number = higher priority
tasks = [
    (3, "Write report"),
    (1, "Fix critical bug"),
    (2, "Refill coffee"),
    (1, "Answer boss's email"),
    (5, "Organize files")
]
# Create a priority queue
pq = tasks.copy()
heapq.heapify(pq)
print("Starting to process tasks...")
print("-" * 20)
while pq:
    # Get the task with the highest priority
    priority, task = heapq.heappop(pq)
    print(f"Processing priority {priority}: {task}")
print("-" * 20)
print("All tasks completed.")

Output:

Starting to process tasks...
--------------------
Processing priority 1: Fix critical bug
Processing priority 1: Answer boss's email
Processing priority 2: Refill coffee
Processing priority 3: Write report
Processing priority 5: Organize files
--------------------
All tasks completed.

Notice how the two tasks with priority 1 were processed in the order they were added to the original list. This is because heapq is a stable queue for items with the same priority.

Handling Complex Data and Custom Priorities

What if your items are more complex objects, or you want to use a different attribute for priority?

a) Using Tuples with Custom Objects

You can create a tuple (priority, object). If two priorities are equal, heapq will look at the next item in the tuple to break the tie. This can cause an error if the items are not comparable (e.g., different types).

Solution: Include a unique counter as the second element in the tuple. This ensures that if two items have the same priority, the one that was added first will be popped first.

import heapq
import time
class Task:
    def __init__(self, description):
        self.description = description
        self.created_at = time.time()
    def __repr__(self):
        return f"Task('{self.description}')"
# Use a counter to ensure stability for items with the same priority
counter = 0
pq = []
task1 = Task("Low priority task")
task2 = Task("High priority task")
task3 = Task("High priority task")
heapq.heappush(pq, (1, counter, task1))
counter += 1
heapq.heappush(pq, (3, counter, task2))
counter += 1
heapq.heappush(pq, (3, counter, task3))
while pq:
    priority, _, task = heapq.heappop(pq)
    print(f"Processing: {task} (Priority: {priority})")
# Output:
# Processing: Task('High priority task') (Priority: 3)
# Processing: Task('High priority task') (Priority: 3)
# Processing: Task('Low priority task') (Priority: 1)

b) Using dataclasses with __lt__

A more "Pythonic" way is to use a dataclass and define the less-than (__lt__) method. This tells heapq how to compare your objects.

import heapq
from dataclasses import dataclass
@dataclass(order=True) # This automatically adds __lt__, __eq__, etc.
class PrioritizedTask:
    priority: int
    description: str
    # The 'order=True' uses the 'priority' field for comparison by default.
# No need for tuples or counters, the class handles it!
pq = []
pq.append(PrioritizedTask(priority=3, description="Write report"))
pq.append(PrioritizedTask(priority=1, description="Fix critical bug"))
pq.append(PrioritizedTask(priority=2, description="Refill coffee"))
heapq.heapify(pq)
while pq:
    task = heapq.heappop(pq)
    print(f"Processing: {task.description} (Priority: {task.priority})")
# Output:
# Processing: Fix critical bug (Priority: 1)
# Processing: Refill coffee (Priority: 2)
# Processing: Write report (Priority: 3)

When to Use Other Options

queue.PriorityQueue

The queue module also provides a PriorityQueue class. It's essentially a thread-safe wrapper around heapq.

Use queue.PriorityQueue when:

• You are working in a multi-threaded application.
• You want an object-oriented interface (.put(), .get()) instead of a functional one (heappush, heappop).

Example:

import queue
import threading
def worker(q):
    while True:
        item = q.get()
        if item is None: # Sentinel value to stop the worker
            break
        print(f"Processing: {item}")
        q.task_done()
# Create a queue
q = queue.PriorityQueue()
# Start a worker thread
threading.Thread(target=worker, args=(q,), daemon=True).start()
# Add items to the queue
q.put((3, "Task 3"))
q.put((1, "Task 1"))
q.put((2, "Task 2