icaoberg / Data structures - Priority Queue

Created Sun, 03 May 2026 00:00:00 +0000 Modified Sun, 03 May 2026 09:38:18 -0400

Jupyter Notebook python

Data structures - Priority Queue¶

A priority queue is an abstract data type where each element has an associated priority. Elements are dequeued in priority order — the element with the highest priority (lowest number by convention) is removed first, regardless of insertion order.

Under the hood this implementation uses Python's heapq module, which maintains a min-heap in O(log n) time.

This notebook covers:

What is a Priority Queue?
The PriorityQueue class
Core operations: push, pop, peek, is_empty, size
Example: hospital emergency triage simulation

1. What is a Priority Queue?¶

Operation	Description	Time complexity
`push(item, priority)`	Add an item with a given priority	O(log n)
`pop()`	Remove and return the highest-priority item	O(log n)
`peek()`	Return the highest-priority item without removing it	O(1)
`is_empty()`	Return `True` if the queue has no items	O(1)
`size()`	Return the number of items	O(1)

Priority convention: lower number = higher priority (priority 1 beats priority 5).
Ties are broken by insertion order so the queue is always stable.

2. The `PriorityQueue` class¶

In [1]:

import heapq
from itertools import count


class PriorityQueue:
    """
    Min-priority queue backed by heapq.

    Lower priority numbers are dequeued first.
    Items with equal priority are returned in insertion order (FIFO).
    """

    def __init__(self):
        self._heap = []          # list of (priority, sequence, item)
        self._counter = count()  # tie-breaker to preserve insertion order

    def push(self, item, priority=0):
        """Add *item* with the given *priority* (lower = higher priority)."""
        heapq.heappush(self._heap, (priority, next(self._counter), item))

    def pop(self):
        """Remove and return the highest-priority item. Raises IndexError if empty."""
        if self.is_empty():
            raise IndexError("pop from an empty priority queue")
        priority, _, item = heapq.heappop(self._heap)
        return item, priority

    def peek(self):
        """Return (item, priority) of the highest-priority element without removing it."""
        if self.is_empty():
            raise IndexError("peek at an empty priority queue")
        priority, _, item = self._heap[0]
        return item, priority

    def is_empty(self):
        """Return True if the queue contains no items."""
        return len(self._heap) == 0

    def size(self):
        """Return the number of items in the queue."""
        return len(self._heap)

    def __repr__(self):
        if self.is_empty():
            return "PriorityQueue(empty)"
        item, pri = self.peek()
        return f"PriorityQueue(next={item!r}, priority={pri}, size={self.size()})"

3. Core operations¶

In [2]:

pq = PriorityQueue()
print("Empty?  ", pq.is_empty())
print("Size:   ", pq.size())
print(repr(pq))

Empty?   True
Size:    0
PriorityQueue(empty)

In [3]:

# push items with different priorities (lower number = higher priority)
tasks = [
    ("send report",   3),
    ("fix outage",    1),
    ("update docs",   5),
    ("code review",   2),
    ("deploy patch",  1),  # same priority as 'fix outage' — inserted later
]

for item, priority in tasks:
    pq.push(item, priority)
    print(f"push({item!r}, priority={priority})  →  {repr(pq)}")

push('send report', priority=3)  →  PriorityQueue(next='send report', priority=3, size=1)
push('fix outage', priority=1)  →  PriorityQueue(next='fix outage', priority=1, size=2)
push('update docs', priority=5)  →  PriorityQueue(next='fix outage', priority=1, size=3)
push('code review', priority=2)  →  PriorityQueue(next='fix outage', priority=1, size=4)
push('deploy patch', priority=1)  →  PriorityQueue(next='fix outage', priority=1, size=5)

In [4]:

item, priority = pq.peek()
print(f"peek()  →  {item!r}  (priority={priority})")
print(f"size()  →  {pq.size()}  (unchanged)")

peek()  →  'fix outage'  (priority=1)
size()  →  5  (unchanged)

In [5]:

# pop all items — they come out in priority order
print(f"{'#':<3} {'Item':<18} {'Priority'}")
print("-" * 32)
rank = 1
while not pq.is_empty():
    item, priority = pq.pop()
    print(f"{rank:<3} {item!r:<18} {priority}")
    rank += 1

#   Item               Priority
--------------------------------
1   'fix outage'       1
2   'deploy patch'     1
3   'code review'      2
4   'send report'      3
5   'update docs'      5

In [6]:

# pop from empty queue raises IndexError
try:
    pq.pop()
except IndexError as e:
    print("Caught:", e)

Caught: pop from an empty priority queue

4. Example: hospital emergency triage simulation¶

In a hospital emergency room, patients are treated by severity, not by arrival time. A priority queue models this naturally: lower triage level = treated first.

Triage level	Severity
1	Immediate — life-threatening
2	Emergent — could deteriorate quickly
3	Urgent — serious but stable
4	Semi-urgent — less urgent
5	Non-urgent — minor complaints

In [7]:

import random
import time

random.seed(7)

TRIAGE_LABELS = {
    1: "Immediate",
    2: "Emergent",
    3: "Urgent",
    4: "Semi-urgent",
    5: "Non-urgent",
}

patients = [
    ("Alice",   3),
    ("Bob",     1),
    ("Carol",   5),
    ("David",   2),
    ("Eve",     1),
    ("Frank",   4),
    ("Grace",   2),
    ("Hector",  3),
]

er = PriorityQueue()

print("Patients arriving at the ER:")
for name, level in patients:
    er.push(name, level)
    print(f"  {name:<8} arrives — triage level {level} ({TRIAGE_LABELS[level]})")

print(f"\nPatients in queue: {er.size()}")
print("\nTreating patients (highest priority first):")
print(f"  {'#':<4} {'Patient':<10} {'Level':<6} {'Severity'}")
print("  " + "-" * 38)

order = 1
while not er.is_empty():
    patient, level = er.pop()
    time.sleep(0.05)  # simulate treatment time
    print(f"  {order:<4} {patient:<10} {level:<6} {TRIAGE_LABELS[level]}")
    order += 1

print("\nAll patients treated.")

Patients arriving at the ER:
  Alice    arrives — triage level 3 (Urgent)
  Bob      arrives — triage level 1 (Immediate)
  Carol    arrives — triage level 5 (Non-urgent)
  David    arrives — triage level 2 (Emergent)
  Eve      arrives — triage level 1 (Immediate)
  Frank    arrives — triage level 4 (Semi-urgent)
  Grace    arrives — triage level 2 (Emergent)
  Hector   arrives — triage level 3 (Urgent)

Patients in queue: 8

Treating patients (highest priority first):
  #    Patient    Level  Severity
  --------------------------------------
  1    Bob        1      Immediate
  2    Eve        1      Immediate
  3    David      2      Emergent
  4    Grace      2      Emergent
  5    Alice      3      Urgent
  6    Hector     3      Urgent
  7    Frank      4      Semi-urgent
  8    Carol      5      Non-urgent

All patients treated.

Licensed under the GNU General Public License v2.0 (GPL-2).
You may redistribute and/or modify this work under the terms of GPL-2.
This work is distributed WITHOUT ANY WARRANTY.
Happy computing.