Python’s heapq module implements a min-heap on top of a plain list. import heapq. heappush: heapq.heappush(heap, item) — O(log n). heappop: heapq.heappop(heap) → smallest item, O(log n). heapify: heapq.heapify(lst) — in-place, O(n). heapreplace: heapq.heapreplace(heap, item) — pop and push, O(log n), more efficient than pop+push. heappushpop: heapq.heappushpop(heap, item) — push then pop, O(log n). nlargest: heapq.nlargest(n, iterable, key=...) — returns sorted list. nsmallest: heapq.nsmallest(n, iterable, key=...) — returns sorted list. merge: heapq.merge(*sorted_iterables, key=..., reverse=False) — lazy O(n log k). max-heap: negate values heapq.heappush(heap, -val). tuple heap: heapq.heappush(heap, (priority, item)) — compare on first element. tie-breaking: (priority, counter, item) — counter avoids comparing uncomparable items. peek: heap[0] — smallest without popping. empty check: while heap:. siftup/siftdown: internal functions, prefer public API. Heap invariant: heap[k] <= heap[2*k+1] and heap[k] <= heap[2*k+2]. Claude Code generates task schedulers, Dijkstra implementations, streaming top-N selectors, and event-driven simulation engines.
CLAUDE.md for heapq
## heapq Stack
- Stdlib: import heapq
- Push: heapq.heappush(heap, (priority, item))
- Pop: priority, item = heapq.heappop(heap)
- Min: heap[0] — peek without removing
- Make: heapq.heapify(list) — O(n) in-place
- Top-N: heapq.nlargest(10, items, key=lambda x: x.score)
- Max-heap: heapq.heappush(heap, -value) — negate for maxheapq Priority Queue Pipeline
# app/heaputil.py — min-heap, max-heap, PriorityQueue, top-N, Dijkstra, median
from __future__ import annotations
import heapq
import itertools
from dataclasses import dataclass, field
from typing import Any, Callable, Generic, Iterable, Iterator, TypeVar
T = TypeVar("T")
K = TypeVar("K")
# ─────────────────────────────────────────────────────────────────────────────
# 1. Priority queue class
# ─────────────────────────────────────────────────────────────────────────────
class MinHeap(Generic[T]):
"""
Min-priority queue with stable ordering on ties via an insertion counter.
Example:
pq = MinHeap[str]()
pq.push("low-priority", priority=10)
pq.push("high-priority", priority=1)
_, task = pq.pop() # "high-priority"
"""
def __init__(self) -> None:
self._heap: list[tuple] = []
self._counter: itertools.count = itertools.count()
def push(self, item: T, priority: int | float = 0) -> None:
"""Push item with given priority (lower = higher urgency)."""
heapq.heappush(self._heap, (priority, next(self._counter), item))
def pop(self) -> tuple[int | float, T]:
"""Pop and return (priority, item) with the smallest priority."""
priority, _, item = heapq.heappop(self._heap)
return priority, item
def peek(self) -> tuple[int | float, T] | None:
"""Return (priority, item) of the smallest item without removing it."""
if not self._heap:
return None
priority, _, item = self._heap[0]
return priority, item
def __len__(self) -> int:
return len(self._heap)
def __bool__(self) -> bool:
return bool(self._heap)
class MaxHeap(Generic[T]):
"""
Max-priority queue (highest priority pops first) via negation.
Example:
pq = MaxHeap[str]()
pq.push("job-a", priority=5)
pq.push("job-b", priority=10)
_, task = pq.pop() # "job-b"
"""
def __init__(self) -> None:
self._heap: list[tuple] = []
self._counter: itertools.count = itertools.count()
def push(self, item: T, priority: int | float = 0) -> None:
heapq.heappush(self._heap, (-priority, next(self._counter), item))
def pop(self) -> tuple[int | float, T]:
neg_priority, _, item = heapq.heappop(self._heap)
return -neg_priority, item
def peek(self) -> tuple[int | float, T] | None:
if not self._heap:
return None
neg_priority, _, item = self._heap[0]
return -neg_priority, item
def __len__(self) -> int:
return len(self._heap)
def __bool__(self) -> bool:
return bool(self._heap)
# ─────────────────────────────────────────────────────────────────────────────
# 2. Top-N helpers
# ─────────────────────────────────────────────────────────────────────────────
def top_n(
items: Iterable[T],
n: int,
key: Callable[[T], Any] = None,
largest: bool = True,
) -> list[T]:
"""
Return the n largest (or smallest) items efficiently.
Uses sort for small n relative to data; O(n log n) otherwise.
Example:
scores = [{"name": "A", "score": 90}, {"name": "B", "score": 75}]
top3 = top_n(scores, 3, key=lambda x: x["score"])
"""
fn = heapq.nlargest if largest else heapq.nsmallest
return fn(n, items, key=key)
def running_top_n(
stream: Iterable[T],
n: int,
key: Callable[[T], Any] | None = None,
) -> list[T]:
"""
Consume a stream, returning the top-n largest seen so far.
O(total * log n) — maintains a window heap of size n.
Example:
result = running_top_n(log_entries, 5, key=lambda e: e.latency_ms)
"""
heap: list = []
_key = key or (lambda x: x)
for item in stream:
k = _key(item)
if len(heap) < n:
heapq.heappush(heap, (k, item))
elif k > heap[0][0]:
heapq.heapreplace(heap, (k, item))
return [item for _, item in heapq.nlargest(n, heap)]
def merge_sorted(*iterables: Iterable[T], key: Callable[[T], Any] | None = None) -> Iterator[T]:
"""
Lazily merge multiple sorted iterables in sorted order.
Example:
combined = list(merge_sorted([1, 3, 5], [2, 4, 6])) # [1, 2, 3, 4, 5, 6]
"""
return heapq.merge(*iterables, key=key)
# ─────────────────────────────────────────────────────────────────────────────
# 3. Running median (two-heap method)
# ─────────────────────────────────────────────────────────────────────────────
class RunningMedian:
"""
Maintain an online running median using two heaps.
Amortized O(log n) per insertion.
Example:
rm = RunningMedian()
for v in [5, 3, 8, 1, 9, 2]:
rm.add(v)
print(rm.median)
"""
def __init__(self) -> None:
self._lo: list[float] = [] # max-heap (negated) — values ≤ median
self._hi: list[float] = [] # min-heap — values ≥ median
def add(self, value: float) -> None:
heapq.heappush(self._lo, -value)
# Balance: move lo's max to hi
heapq.heappush(self._hi, -heapq.heappop(self._lo))
# Keep lo at least as large as hi
if len(self._lo) < len(self._hi):
heapq.heappush(self._lo, -heapq.heappop(self._hi))
@property
def median(self) -> float:
if not self._lo:
raise ValueError("No values")
if len(self._lo) > len(self._hi):
return float(-self._lo[0])
return (-self._lo[0] + self._hi[0]) / 2.0
def __len__(self) -> int:
return len(self._lo) + len(self._hi)
# ─────────────────────────────────────────────────────────────────────────────
# 4. Dijkstra shortest path
# ─────────────────────────────────────────────────────────────────────────────
def dijkstra(
graph: dict[str, list[tuple[str, float]]],
start: str,
end: str | None = None,
) -> dict[str, float]:
"""
Dijkstra's shortest-path algorithm using heapq.
graph: {node: [(neighbor, weight), ...]}
Returns distances from start to all reachable nodes.
Example:
graph = {
"A": [("B", 1), ("C", 4)],
"B": [("C", 2), ("D", 5)],
"C": [("D", 1)],
"D": [],
}
dist = dijkstra(graph, "A")
print(dist["D"]) # 4.0
"""
dist: dict[str, float] = {}
heap: list[tuple[float, str]] = [(0.0, start)]
while heap:
d, node = heapq.heappop(heap)
if node in dist:
continue
dist[node] = d
if node == end:
break
for neighbor, weight in graph.get(node, []):
if neighbor not in dist:
heapq.heappush(heap, (d + weight, neighbor))
return dist
def shortest_path(
graph: dict[str, list[tuple[str, float]]],
start: str,
end: str,
) -> tuple[float, list[str]]:
"""
Return (distance, path) for the shortest path from start to end.
Example:
dist, path = shortest_path(graph, "A", "D")
print(f"{dist}: {' → '.join(path)}")
"""
dist: dict[str, float] = {}
prev: dict[str, str] = {}
heap: list[tuple[float, str]] = [(0.0, start)]
while heap:
d, node = heapq.heappop(heap)
if node in dist:
continue
dist[node] = d
for neighbor, weight in graph.get(node, []):
if neighbor not in dist:
nd = d + weight
if nd < dist.get(neighbor, float("inf")):
prev[neighbor] = node
heapq.heappush(heap, (nd, neighbor))
# Reconstruct path
path = []
cur = end
while cur in prev:
path.append(cur)
cur = prev[cur]
path.append(start)
path.reverse()
return dist.get(end, float("inf")), path
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("=== heapq demo ===")
print("\n--- MinHeap ---")
pq: MinHeap[str] = MinHeap()
for task, pri in [("low", 10), ("high", 1), ("medium", 5), ("urgent", 0)]:
pq.push(task, priority=pri)
print(f" peek: {pq.peek()}")
while pq:
pri, task = pq.pop()
print(f" pop: {pri:3d} → {task!r}")
print("\n--- MaxHeap ---")
mh: MaxHeap[str] = MaxHeap()
for name, score in [("alice", 85), ("bob", 92), ("carol", 78)]:
mh.push(name, priority=score)
while mh:
score, name = mh.pop()
print(f" {score}: {name}")
print("\n--- top_n ---")
data = [{"name": chr(65 + i), "score": (i * 37) % 100} for i in range(10)]
top3 = top_n(data, 3, key=lambda x: x["score"])
print(f" top 3: {[(d['name'], d['score']) for d in top3]}")
bot3 = top_n(data, 3, key=lambda x: x["score"], largest=False)
print(f" bot 3: {[(d['name'], d['score']) for d in bot3]}")
print("\n--- running_top_n ---")
result = running_top_n(range(1000), 5, key=lambda x: (x * 31) % 1000)
print(f" top 5 (hash): {result}")
print("\n--- merge_sorted ---")
merged = list(merge_sorted([1, 3, 5], [2, 4, 6], [0, 7, 8]))
print(f" {merged}")
print("\n--- RunningMedian ---")
rm = RunningMedian()
for v in [5, 3, 8, 1, 9, 2, 7]:
rm.add(v)
print(f" added {v:3d} → median={rm.median}")
print("\n--- dijkstra ---")
graph = {
"A": [("B", 1), ("C", 4)],
"B": [("C", 2), ("D", 5)],
"C": [("D", 1)],
"D": [],
}
distances = dijkstra(graph, "A")
print(f" distances from A: {distances}")
print("\n--- shortest_path ---")
dist, path = shortest_path(graph, "A", "D")
print(f" A → D: {dist:.1f} path: {' → '.join(path)}")
print("\n=== done ===")For the sortedcontainers.SortedList alternative — sortedcontainers.SortedList (PyPI) maintains sorted order for all insertions and supports O(log n) bisect_left/bisect_right and O(1) indexed access; heapq only guarantees O(1) access to the minimum, not arbitrary-rank access — use SortedList when you need sorted iteration, rank queries, or range slicing alongside insertions and deletions, heapq for pure priority-queue workloads (Dijkstra, simulation event queues) where only peek/pop-minimum matters. For the queue.PriorityQueue alternative — queue.PriorityQueue wraps heapq with thread-safe locking and a get(block=True, timeout=N) API, making it suitable for multi-threaded producer-consumer patterns; heapq is single-threaded only — use queue.PriorityQueue in multithreaded code where multiple threads push and pop jobs, heapq in single-threaded algorithms, async code (use asyncio.PriorityQueue), or when you need direct list access. The Claude Skills 360 bundle includes heapq skill sets covering MinHeap/MaxHeap generic classes with push/pop/peek, top_n()/running_top_n() top-N selectors with streaming support, merge_sorted() lazy sorted-merge, RunningMedian two-heap online median, and dijkstra()/shortest_path() graph shortest-path implementations. Start with the free tier to try priority queue patterns and heapq algorithm code generation.