Binary Search with Bisect and Priority Queues with Heapq
Binary Search with Bisect and Priority Queues with Heapq
After reading this chapter, you will master Python’s algorithmic collection tools: maintain sorted sequences in \(O(\log N)\) search time using bisect, implement priority-driven task schedulers with heapq, compute top-\(K\) extremes efficiently, and prevent tuple tie-breaker comparison crashes.
Mental model
Sorting an entire list costs \(O(N \log N)\). When elements arrive dynamically over time, re-sorting the list after each insertion quickly degrades system performance. Python provides two specialized modules in the standard library to solve this:
1. bisect: Bisection and Binary Search
The bisect module implements binary search over sorted sequences. It locates the insertion index for a new item in \(O(\log N)\) time, and inserts it with insort() while preserving sorted order:
Sorted List: [ 10, 25, 50, 75, 90 ]
▲ ▲
Search for 30: │ │ 30 belongs at index 2 (between 25 and 50)
└────────┴── Found in O(log N) comparisons
2. heapq: Min-Heap Priority Queues
The heapq module implements a binary min-heap using a standard Python list. In a min-heap, heap[0] is always the smallest element, and every parent node at index \(k\) satisfies:
\[\text{heap}[k] \le \text{heap}[2k + 1] \quad \text{and} \quad \text{heap}[k] \le \text{heap}[2k + 2]\]
Binary Tree View: Contiguous List Representation:
[ 10 ] [ 10, 20, 15, 30, 40, 50, 60 ]
/ \ ▲ ▲ ▲ ▲ ▲ ▲ ▲
[ 20 ] [ 15 ] 0 1 2 3 4 5 6
/ \ / \
[30] [40] [50] [60]
- Push (
heappush): Adds an item and restores heap invariants in \(O(\log N)\). - Pop (
heappop): Removes and returns the smallest element in \(O(\log N)\). - Peek (
heap[0]): Inspects the minimum element in instantaneous \(O(1)\) time.
Minimal example
Save as algorithms_demo.py:
# algorithms_demo.py
import bisect
import heapq
def main() -> None:
# 1. Bisect: Maintain sorted latency samples
latencies = [12.4, 25.1, 45.0, 88.2, 120.5]
new_sample = 32.8
# Find index and insert while preserving sort
idx = bisect.bisect_left(latencies, new_sample)
print(f"Sample {new_sample}ms belongs at index: {idx}")
bisect.insort(latencies, new_sample)
print(f"Updated sorted latencies: {latencies}")
# 2. Heapq: Priority job queue
job_queue: list[tuple[int, str]] = []
# Items: (priority_number, job_name) -- lowest number = highest priority
heapq.heappush(job_queue, (3, "routine_log_cleanup"))
heapq.heappush(job_queue, (1, "SECURITY_PATCH_RESTART"))
heapq.heappush(job_queue, (2, "customer_invoice_generation"))
print("\nProcessing jobs in priority order:")
while job_queue:
priority, job = heapq.heappop(job_queue)
print(f" [Priority {priority}] Executing: {job}")
if __name__ == "__main__":
main()Run via uv run python algorithms_demo.py:
Sample 32.8ms belongs at index: 2
Updated sorted latencies: [12.4, 25.1, 32.8, 45.0, 88.2, 120.5]
Processing jobs in priority order:
[Priority 1] Executing: SECURITY_PATCH_RESTART
[Priority 2] Executing: customer_invoice_generation
[Priority 3] Executing: routine_log_cleanup
Worked examples
Case 1: Numeric Score / Severity Bucketing with bisect
Instead of writing verbose chains of if/elif statements to map numeric thresholds to categorical levels (e.g. alert severity, HTTP response time tiers), bisect performs the classification in a single binary search lookup:
# alert_classifier.py
import bisect
def classify_severity(metric_value: float) -> str:
"""Classify CPU utilization into alert severity tiers using binary search."""
# Breakpoints must be strictly sorted
thresholds = [50.0, 75.0, 90.0]
severity_labels = ["INFO", "WARNING", "CRITICAL", "EMERGENCY"]
# bisect_right returns the index where metric_value falls
index = bisect.bisect_right(thresholds, metric_value)
return severity_labels[index]
def main() -> None:
test_values = [35.0, 50.0, 68.2, 75.0, 88.0, 91.5, 99.9]
print("Alert Level Classification:")
for val in test_values:
level = classify_severity(val)
print(f" CPU: {val:5.1f}% -> Alert: {level}")
if __name__ == "__main__":
main()Run:
uv run python alert_classifier.pyOutput:
Alert Level Classification:
CPU: 35.0% -> Alert: INFO
CPU: 50.0% -> Alert: WARNING
CPU: 68.2% -> Alert: WARNING
CPU: 75.0% -> Alert: CRITICAL
CPU: 88.0% -> Alert: CRITICAL
CPU: 91.5% -> Alert: EMERGENCY
CPU: 99.9% -> Alert: EMERGENCY
Case 2: Multi-Criteria Priority Scheduler with Tie-Breaking Counter
When pushing tasks into a heap as tuples (priority, task_object), if two tasks share identical priority numbers, Python attempts to compare the task_object instances with <. If the objects do not implement __lt__, Python raises TypeError. A monotonic counter resolves ties safely:
# priority_scheduler.py
import heapq
import itertools
from typing import NamedTuple
class Task(NamedTuple):
name: str
payload: dict[str, str]
class PriorityScheduler:
def __init__(self) -> None:
self._heap: list[tuple[int, int, Task]] = []
self._counter = itertools.count() # Unique tie-breaker counter
def add_task(self, task: Task, priority: int) -> None:
count = next(self._counter)
# Tuple ordering: priority first; if equal, count breaks the tie!
heapq.heappush(self._heap, (priority, count, task))
def pop_task(self) -> Task:
if not self._heap:
raise IndexError("Scheduler is empty")
priority, count, task = heapq.heappop(self._heap)
return task
def __len__(self) -> int:
return len(self._heap)
def main() -> None:
scheduler = PriorityScheduler()
# Two tasks with IDENTICAL priority (1)
scheduler.add_task(Task("deploy_auth", {"env": "prod"}), priority=1)
scheduler.add_task(Task("flush_cache", {"env": "prod"}), priority=1)
scheduler.add_task(Task("send_digest", {"freq": "daily"}), priority=3)
print("Dispatching scheduled tasks in FIFO-priority order:")
while len(scheduler) > 0:
task = scheduler.pop_task()
print(f" Dispatched: {task.name} ({task.payload})")
if __name__ == "__main__":
main()Run:
uv run python priority_scheduler.pyOutput:
Dispatching scheduled tasks in FIFO-priority order:
Dispatched: deploy_auth ({'env': 'prod'})
Dispatched: flush_cache ({'env': 'prod'})
Dispatched: send_digest ({'freq': 'daily'})
Notice: Tasks with identical priority are dispatched in exact First-In, First-Out (FIFO) arrival order!
Case 3: Finding Top \(K\) Extremes with nlargest and nsmallest
When querying the top 5 highest memory-consuming processes out of 100,000 servers, sorting the entire list requires \(O(N \log N)\) time and allocates a full new array. heapq.nlargest maintains an internal bounded heap of size \(K\), executing in optimal \(O(N \log K)\) time:
# top_k_metrics.py
import heapq
def main() -> None:
fleet_metrics = [
{"host": "worker-01", "mem_mb": 14200},
{"host": "worker-02", "mem_mb": 31500},
{"host": "worker-03", "mem_mb": 8400},
{"host": "worker-04", "mem_mb": 62100},
{"host": "worker-05", "mem_mb": 45000},
{"host": "worker-06", "mem_mb": 18900},
]
# Find the top 3 highest memory consumers in O(N log K)
top_consumers = heapq.nlargest(3, fleet_metrics, key=lambda x: x["mem_mb"])
print("Top 3 Highest Memory Consumers:")
for rank, node in enumerate(top_consumers, start=1):
print(f" #{rank}: {node['host']} consuming {node['mem_mb']:,} MB")
if __name__ == "__main__":
main()Run:
uv run python top_k_metrics.pyOutput:
Top 3 Highest Memory Consumers:
#1: worker-04 consuming 62,100 MB
#2: worker-05 consuming 45,000 MB
#3: worker-02 consuming 31,500 MB
Pitfalls
Pitfall 1: Calling bisect on an Unsorted List
The bisection algorithm assumes the underlying sequence is already sorted. Calling bisect on an unsorted collection produces completely incorrect indices without raising any warning or exception:
# THE TRAP:
unsorted_list = [50, 10, 90, 20]
# bisect assumes sorted order! Binary search branches unpredictably!
bad_index = bisect.bisect_left(unsorted_list, 15)
# THE FIX: Ensure sequence is sorted before querying
sorted_list = sorted(unsorted_list)
safe_index = bisect.bisect_left(sorted_list, 15)
print(f"Correct index in sorted list: {safe_index}")Pitfall 2: Modifying Elements Inside a Heap Directly
heapq functions maintain the heap invariant on append and pop. If you mutate an element’s priority while it is inside the list, the heap becomes corrupted:
# THE TRAP:
heap = [10, 20, 30]
heapq.heapify(heap)
heap[1] = 5 # Corrupts the heap! heap[0] is 10, but heap[1] is now smaller (5)!
print(f"Corrupted minimum: {heapq.heappop(heap)}") # Returns 10, NOT 5!# THE FIX: Call heapq.heapify() after modifying in-place, or remove and re-push
heap = [10, 20, 30]
heapq.heapify(heap)
heap[1] = 5
heapq.heapify(heap) # Restores heap invariant in O(N)
print(f"Restored minimum: {heapq.heappop(heap)}") # Returns 5!Exercises
- Given a list of timestamps, use
bisect.bisect_leftto find the earliest event that occurred after a given target time. - Build an event scheduler that takes tuples of
(run_at_timestamp, callback_func)and executes them in temporal order usingheapq. - Given a stream of numbers, maintain a sliding window of the 5 largest numbers seen so far using
heapq. - Measure the speed difference between
heapq.nlargest(10, data)versussorted(data, reverse=True)[:10]on a dataset of 1,000,000 floats. - Demonstrate how
heapq.merge(*sorted_iterables)can merge three sorted log streams into a single sorted output stream with minimal memory overhead.
Further reading
- Python Documentation:
bisect— Array bisection algorithm. - Python Documentation:
heapq— Heap queue algorithm. - David Beazley & Brian K. Jones: Python Cookbook (Chapter 1: Data Structures and Algorithms).