A priority queue serves elements by priority instead of arrival order, and in Python we build one with the heapq module on a plain list, or with queue.PriorityQueue when several threads share the queue; both always return the element with the lowest priority value first. Job schedulers and shortest-path algorithms rely on it.
import heapq
tasks = []
heapq.heappush(tasks, (2, "send report"))
heapq.heappush(tasks, (1, "fix outage"))
first = heapq.heappop(tasks) # (1, 'fix outage')We use heapq with tuples, add a counter for stable ties, build a max-heap with the new Python 3.14 functions, share a queue between threads, and compare with the bisect approach. The examples run on Python 3.14.
1. Implementing Priority Queue using heapq Module
The heapq module keeps a list in heap order, where heap[0] is always the smallest element. Pushing and popping take O(log n) time, and heapify() turns an existing list into a heap in O(n).
jobs = [(5, "backup"), (1, "deploy"), (3, "email")]
heapq.heapify(jobs)
order = [heapq.heappop(jobs)[1] for _ in range(3)] # ['deploy', 'email', 'backup']Tuples compare by their first element, so the priority goes first. Only heap[0] is guaranteed to be the smallest; the rest of the list is not sorted.
2. Ties and the Counter Tie-Breaker
When two priorities tie, Python compares the next tuple element, so equal priorities do not leave in arrival order. If that element cannot be compared, such as a dictionary, the push raises TypeError. A running counter as the second field fixes both problems.
import itertools
counter = itertools.count()
heap = []
for prio, task in [(2, {"name": "a"}), (1, {"name": "b"}), (2, {"name": "c"})]:
heapq.heappush(heap, (prio, next(counter), task))
names = [heapq.heappop(heap)[2]["name"] for _ in range(3)] # ['b', 'a', 'c']The counter makes equal priorities leave in the order they arrived, and Python never reaches the dictionaries when comparing.
3. Max-Heap: Highest Priority First
Python 3.14 added max-heap functions to heapq, such as heappush_max() and heappop_max(). On older versions, the common trick is to push negated numbers, which works only for numeric priorities.
scores = []
for s in [40, 95, 70]:
heapq.heappush_max(scores, s)
best = heapq.heappop_max(scores) # 95 (Python 3.14+)
old_style = []
for s in [40, 95, 70]:
heapq.heappush(old_style, -s)
best_old = -heapq.heappop(old_style) # 954. Thread-Safe Priority Queue with queue.PriorityQueue
The queue.PriorityQueue class wraps a heap with locks, so producer and consumer threads can share it safely, and get() blocks until an element is available. For items that cannot be compared, a dataclass with order=True generates comparison methods, and field(compare=False) leaves the item out of them, so only the priority is compared.
from dataclasses import dataclass, field
from queue import PriorityQueue
from typing import Any
@dataclass(order=True)
class Job:
priority: int
item: Any = field(compare=False)
q = PriorityQueue()
for p, name in [(5, "how"), (4, "to"), (1, "do"), (3, "in"), (2, "java")]:
q.put(Job(p, name))
served = [q.get().item for _ in range(5)] # ['do', 'java', 'in', 'to', 'how']For single-threaded code, the locking only adds overhead, so plain heapq is faster there. Since Python 3.13, q.shutdown() stops waiting consumers cleanly.
5. Implementing Priority Queue using bisect Module
The bisect.insort() function inserts an element into a sorted list at the right position. Finding the position takes O(log n), but inserting shifts the elements behind it, so each insert costs O(n); the advantage is that the whole list stays sorted and can be printed in order.
import bisect
ordered = []
for job in [(3, "in"), (1, "do"), (2, "java")]:
bisect.insort(ordered, job)
# ordered = [(1, 'do'), (2, 'java'), (3, 'in')]
lowest = ordered.pop(0) # (1, 'do')| Approach | Push | Pop | Best for |
|---|---|---|---|
| heapq | O(log n) | O(log n) | Most single-threaded code |
| queue.PriorityQueue | O(log n) plus locking | O(log n) plus locking | Several threads |
| bisect.insort | O(n) | O(n) from the front | Small queues that must stay fully sorted |
6. A Simple Task Scheduler
A scheduler pops the most urgent task and can add new tasks while it works. With the counter from section 2, tasks of equal priority run in arrival order.
def run(initial):
heap, seq = [], itertools.count()
for prio, name in initial:
heapq.heappush(heap, (prio, next(seq), name))
done = []
while heap:
prio, _, name = heapq.heappop(heap)
done.append(name)
if name == "deploy": # deploying creates a follow-up task
heapq.heappush(heap, (1, next(seq), "smoke test"))
return done
log = run([(2, "deploy"), (3, "email"), (2, "backup")])
# ['deploy', 'smoke test', 'backup', 'email']7. Priority Queue FAQs
A heap has no update or peek method like other queue libraries, so these tasks need small workarounds.
7.1. How do we change the priority of a queued item?
A heap has no update operation. The usual approach from the heapq docs marks the old entry as removed, pushes a new entry, and skips removed entries when popping.
7.2. How do we look at the next item without removing it?
We read heap[0], which is always the smallest element. A queue.PriorityQueue has no peek method, because another thread could take the item at any moment.
7.3. Is heapq.nsmallest() a priority queue?
No, it is a one-off query. heapq.nsmallest(3, data) returns the three smallest items of a collection, which is handy for top-N reports, without keeping a heap.
8. Conclusion
The heapq module turns a list into a fast priority queue, with a counter as the tie-breaker and the Python 3.14 *_max functions for highest-first order. queue.PriorityQueue adds the locking that threads need, and bisect.insort() suits small queues that must stay sorted.
9. References
Happy Learning !!
