Binary Search with Bisect and Priority Queues with Heapq

Updated

September 7, 2026

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:

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.py

Output:

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.py

Output:

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.py

Output:

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

  1. Given a list of timestamps, use bisect.bisect_left to find the earliest event that occurred after a given target time.
  2. Build an event scheduler that takes tuples of (run_at_timestamp, callback_func) and executes them in temporal order using heapq.
  3. Given a stream of numbers, maintain a sliding window of the 5 largest numbers seen so far using heapq.
  4. Measure the speed difference between heapq.nlargest(10, data) versus sorted(data, reverse=True)[:10] on a dataset of 1,000,000 floats.
  5. 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: bisectArray bisection algorithm.
  • Python Documentation: heapqHeap queue algorithm.
  • David Beazley & Brian K. Jones: Python Cookbook (Chapter 1: Data Structures and Algorithms).