A priority queue is a data structure where elements are dequeued based on priority rather than insertion order. Python provides several ways to implement a priority queue.
heapq (Efficient and Built-in)Python's heapq module implements a min-heap by default, where the smallest element has the highest priority.
import heapq
# Create an empty priority queue
pq = []
# Add elements with heapq.heappush()
heapq.heappush(pq, 3)
heapq.heappush(pq, 1) # Lowest priority (highest priority)
heapq.heappush(pq, 4)
heapq.heappush(pq, 2)
print(pq) # Output: [1, 2, 4, 3] (Heap structure, not sorted)
# Remove elements with heapq.heappop()
print(heapq.heappop(pq)) # Output: 1 (Smallest element)
print(heapq.heappop(pq)) # Output: 2
* Efficient: heappush() and heappop() run in O(log n) time.
* Always dequeues the smallest element first (Min-Heap).
Since heapq is a min-heap by default, we can simulate a max-heap by pushing negative values.
import heapq
pq = []
heapq.heappush(pq, -3)
heapq.heappush(pq, -1)
heapq.heappush(pq, -4)
heapq.heappush(pq, -2)
print(-heapq.heappop(pq)) # Output: 4 (Highest priority first)
print(-heapq.heappop(pq)) # Output: 3
* Largest element dequeued first.
* Negating values is a common trick for max-heaps.
heapq with Tuples for Custom PriorityIf we need custom priority values, we can store elements as tuples (priority, item).
import heapq
pq = []
heapq.heappush(pq, (2, "Task B")) # Lower number = higher priority
heapq.heappush(pq, (1, "Task A"))
heapq.heappush(pq, (3, "Task C"))
print(heapq.heappop(pq)) # Output: (1, 'Task A')
print(heapq.heappop(pq)) # Output: (2, 'Task B')
* Maintains a priority order based on the first tuple element.
* Useful for scheduling tasks or event processing.
queue.PriorityQueue (Thread-Safe)The queue.PriorityQueue class provides a thread-safe priority queue.
from queue import PriorityQueue
pq = PriorityQueue()
pq.put((2, "Task B"))
pq.put((1, "Task A"))
pq.put((3, "Task C"))
print(pq.get()) # Output: (1, 'Task A')
print(pq.get()) # Output: (2, 'Task B')
* Thread-safe for multi-threaded applications.
* Uses heapq internally, so operations are O(log n).
If we want more control, we can implement a priority queue class using heapq.
import heapq
class PriorityQueue:
def __init__(self):
self.queue = []
def push(self, priority, item):
heapq.heappush(self.queue, (priority, item))
def pop(self):
return heapq.heappop(self.queue)[1] # Return only the item
def is_empty(self):
return len(self.queue) == 0
# Example usage
pq = PriorityQueue()
pq.push(2, "Task B")
pq.push(1, "Task A")
pq.push(3, "Task C")
print(pq.pop()) # Output: 'Task A' (Highest priority)
print(pq.pop()) # Output: 'Task B'
* Encapsulates priority queue behavior.
* Cleaner and reusable code.
| Method | Pros | Cons |
|---|---|---|
heapq |
Fast (O(log n)), Simple | Not thread-safe |
heapq with tuples |
Custom priority support | More manual handling |
queue.PriorityQueue |
Thread-safe, Easy to use | Slightly slower |
| Custom Class | More control, Readable | Requires implementation |