Python heapq Module Explained — Min-Heaps, Priority Queues and Real-World Patterns

import heapq

# Start with a plain Python list — heapq works directly on lists
task_priorities = []

# heappush maintains the heap property after every insertion
heapq.heappush(task_priorities, 5)   # low priority task
heapq.heappush(task_priorities, 1)   # critical task
heapq.heappush(task_priorities, 3)   # medium priority task
heapq.heappush(task_priorities, 2)   # high priority task
heapq.heappush(task_priorities, 4)   # low-ish priority task

# The raw list is NOT fully sorted — it only guarantees index 0 is smallest
print("Raw heap list:", task_priorities)

# index 0 is always the minimum — O(1) peek, no searching needed
print("Most urgent task priority:", task_priorities[0])

# heappop removes and returns the smallest item, then restores heap property
most_urgent = heapq.heappop(task_priorities)
print("Popped (processed) priority:", most_urgent)
print("Heap after pop:", task_priorities)

# heapify turns an existing list into a heap in-place — O(n), not O(n log n)
unsorted_scores = [42, 7, 99, 15, 3, 61]
heapq.heapify(unsorted_scores)
print("Heapified list:", unsorted_scores)  # smallest is at index 0

import heapq
from dataclasses import dataclass
from itertools import count

@dataclass
class SupportTicket:
    ticket_id: str
    description: str
    # Note: no __lt__ defined — Python can't compare these directly

class PriorityQueue:
    """
    A min-heap backed priority queue that safely handles
    custom objects and tie-breaking between equal priorities.
    """
    def __init__(self):
        self._heap = []                  # the underlying list heapq operates on
        self._counter = count()          # unique sequence number per push

    def push(self, priority: int, item: object) -> None:
        # Tuple layout: (priority, tie_breaker, item)
        # tie_breaker ensures we never compare 'item' objects directly
        tie_breaker = next(self._counter)
        heapq.heappush(self._heap, (priority, tie_breaker, item))

    def pop(self) -> object:
        if not self._heap:
            raise IndexError("pop from an empty priority queue")
        priority, _, item = heapq.heappop(self._heap)  # unpack, discard tie_breaker
        return priority, item

    def peek(self) -> object:
        """Look at the highest-priority item without removing it — O(1)"""
        if not self._heap:
            raise IndexError("peek at an empty priority queue")
        priority, _, item = self._heap[0]  # index 0 is always the minimum
        return priority, item

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


# --- Demo: simulating a customer support ticket system ---
queue = PriorityQueue()

# Lower number = higher urgency (1 is critical, 5 is low)
queue.push(3, SupportTicket("T-001", "Password reset request"))
queue.push(1, SupportTicket("T-002", "Production database is down!"))
queue.push(1, SupportTicket("T-003", "CEO can't log in"))   # same priority as T-002
queue.push(5, SupportTicket("T-004", "Update billing address"))
queue.push(2, SupportTicket("T-005", "App crashing for 10% of users"))

print(f"Queue has {len(queue)} tickets\n")
print("Processing tickets in priority order:")

while queue:
    priority, ticket = queue.pop()
    print(f"  [Priority {priority}] {ticket.ticket_id}: {ticket.description}")

import heapq

# Simulated game session data — player name and their score
game_sessions = [
    {"player": "alex_torres",    "score": 14_820},
    {"player": "priya_nair",     "score": 31_500},
    {"player": "sam_kowalski",   "score": 9_740},
    {"player": "mia_chen",       "score": 28_990},
    {"player": "jordan_okafor",  "score": 45_200},
    {"player": "luna_garcia",    "score": 38_450},
    {"player": "felix_braun",    "score": 22_100},
    {"player": "isabel_santos",  "score": 50_010},
    {"player": "kai_nakamura",   "score": 17_300},
    {"player": "chloe_martin",   "score": 41_880},
    {"player": "omar_hassan",    "score": 5_620},
    {"player": "ruby_patel",     "score": 33_760},
]

# Get the top 5 players — key= works just like sorted()'s key parameter
top_5 = heapq.nlargest(5, game_sessions, key=lambda session: session["score"])

print("🏆 Top 5 Leaderboard:")
for rank, session in enumerate(top_5, start=1):
    print(f"  #{rank}  {session['player']:<20} {session['score']:>7,} pts")

print()

# Get the 3 lowest scorers — useful for 'needs improvement' analytics
bottom_3 = heapq.nsmallest(3, game_sessions, key=lambda session: session["score"])

print("📈 Players who might need a tutorial:")
for session in bottom_3:
    print(f"  {session['player']:<20} {session['score']:>7,} pts")

import heapq

# ── Pattern 1: Negation trick for numeric priorities ──────────────────────────
print("=== Max-Heap via Negation ===")

bandwidth_requests = []   # each item represents a download request size in MB

# Push NEGATED values so the largest becomes the 'smallest' in the min-heap
for request_size_mb in [120, 500, 45, 800, 300, 75]:
    heapq.heappush(bandwidth_requests, -request_size_mb)  # negate on push

# Pop gives us the most negative = the original largest value
while bandwidth_requests:
    largest_request = -heapq.heappop(bandwidth_requests)  # negate on pop to recover
    print(f"  Processing request: {largest_request} MB")

print()

# ── Pattern 2: Wrapper class for readable max-heap with objects ───────────────
print("=== Max-Heap via Wrapper Class ===")

from dataclasses import dataclass, field

@dataclass
class MaxHeapItem:
    """Wraps any comparable value and flips comparison to create max-heap behaviour."""
    priority: int
    label: str

    def __lt__(self, other: "MaxHeapItem") -> bool:
        # Reversed comparison — higher priority wins the 'smallest' slot
        return self.priority > other.priority

job_queue = []
heapq.heappush(job_queue, MaxHeapItem(priority=3,  label="Generate monthly report"))
heapq.heappush(job_queue, MaxHeapItem(priority=10, label="Send payment confirmation"))
heapq.heappush(job_queue, MaxHeapItem(priority=7,  label="Resize uploaded images"))
heapq.heappush(job_queue, MaxHeapItem(priority=1,  label="Archive old log files"))

print("Jobs processed highest-priority first:")
while job_queue:
    job = heapq.heappop(job_queue)
    print(f"  [Priority {job.priority:>2}] {job.label}")