Specialized Collections, Enums, and Heaps

Updated

September 7, 2026

Specialized Collections, Enums, and Heaps

After reading this chapter, you will master the high-performance data structures in Python’s standard library: automatic default grouping with defaultdict, tallying with Counter, bounded FIFO queues with deque, lightweight tuple records with namedtuple, strict domain constants with enum.Enum, min-heap priority queues with heapq, and logarithmic search with bisect.

Mental model

While list and dict handle 80% of common programming tasks, high-throughput systems require purpose-built data structures to prevent performance bottlenecks:

Standard Library Collection Selector:
  Need O(1) double-ended appends/pops with max capacity? ──▶ collections.deque
  Need automatic fallback values for missing dict keys?  ──▶ collections.defaultdict
  Need frequency tallies and multiset arithmetic?         ──▶ collections.Counter
  Need type-safe constant enumerations?                   ──▶ enum.Enum / IntEnum
  Need O(log N) priority queue extraction (min/max)?     ──▶ heapq (binary heap)
  Need O(log N) insertion into already-sorted sequences?  ──▶ bisect (binary search)

Minimal example

Save as collections_showcase.py:

# collections_showcase.py
from collections import Counter, defaultdict, deque
import heapq
from enum import Enum

class NodeStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    OFFLINE = "offline"

def main() -> None:
    # 1. Counter: Multiset frequency analysis
    status_events = ["healthy", "healthy", "degraded", "healthy", "offline", "degraded"]
    counts = Counter(status_events)
    print(f"Status distribution: {counts}")
    print(f"Top 2 frequent states: {counts.most_common(2)}")

    # 2. defaultdict: Grouping without boilerplate
    cluster_nodes = defaultdict(list)
    cluster_nodes["us-east"].append("node-01")
    cluster_nodes["us-east"].append("node-02")
    cluster_nodes["eu-west"].append("node-10")
    print(f"\nGrouped cluster nodes: {dict(cluster_nodes)}")

    # 3. deque: Fixed-size sliding window (drops oldest items automatically)
    rolling_metrics = deque(maxlen=3)
    for sample in [10.5, 12.1, 14.8, 19.2, 8.4]:
        rolling_metrics.append(sample)
    print(f"\nLast 3 metrics in sliding window: {list(rolling_metrics)}")

    # 4. heapq: Min-Heap priority queue
    task_queue: list[tuple[int, str]] = []
    heapq.heappush(task_queue, (5, "Log rotation"))
    heapq.heappush(task_queue, (1, "Power outage failover"))
    heapq.heappush(task_queue, (3, "Security scan"))

    highest_priority = heapq.heappop(task_queue)
    print(f"\nExtracted highest priority task (min value): {highest_priority}")

if __name__ == "__main__":
    main()

Run via uv run python collections_showcase.py:

Status distribution: Counter({'healthy': 3, 'degraded': 2, 'offline': 1})
Top 2 frequent states: [('healthy', 3), ('degraded', 2)]

Grouped cluster nodes: {'us-east': ['node-01', 'node-02'], 'eu-west': ['node-10']}

Last 3 metrics in sliding window: [14.8, 19.2, 8.4]

Extracted highest priority task (min value): (1, 'Power outage failover')

Worked examples

Case 1: High-Speed Counter Arithmetic for Anomaly Detection

Counter supports multiset mathematical operations (+, -, & intersection, | union), making it ideal for detecting sudden traffic surges or baseline deviations:

# traffic_anomaly_counter.py
from collections import Counter

def detect_traffic_anomalies(baseline_ips: list[str], active_ips: list[str]) -> None:
    baseline = Counter(baseline_ips)
    active = Counter(active_ips)

    # Counter subtraction: active - baseline leaves only surging counts
    surge = active - baseline
    print("Traffic surge over baseline:")
    for ip, extra_requests in surge.most_common():
        print(f"  IP: {ip:15} -> {extra_requests} surge requests")

if __name__ == "__main__":
    normal_window = ["192.168.1.10"] * 5 + ["10.0.0.2"] * 10
    spike_window  = ["192.168.1.10"] * 25 + ["10.0.0.2"] * 12 + ["172.16.0.99"] * 50

    detect_traffic_anomalies(normal_window, spike_window)

Run:

uv run python traffic_anomaly_counter.py

Output:

Traffic surge over baseline:
  IP: 172.16.0.99     -> 50 surge requests
  IP: 192.168.1.10    -> 20 surge requests
  IP: 10.0.0.2        -> 2 surge requests

Case 2: Logarithmic Table Lookup with bisect

Searching a list with linear scan is \(O(N)\). For already-sorted lists (e.g. latency SLA tiers or timestamp index maps), bisect.bisect_right() achieves \(O(\log N)\) performance:

# sla_bisect_lookup.py
import bisect

def classify_sla_tier(latency_ms: float) -> str:
    # Latency thresholds (must be sorted!)
    thresholds = [10.0, 50.0, 200.0, 1000.0]
    tiers = ["PLATINUM", "GOLD", "SILVER", "BRONZE", "BREACHED"]

    # bisect_right finds the insertion point in O(log N) time
    idx = bisect.bisect_right(thresholds, latency_ms)
    return tiers[idx]

if __name__ == "__main__":
    latencies = [4.2, 10.0, 25.0, 75.0, 450.0, 2500.0]
    for lat in latencies:
        print(f"Latency: {lat:6.1f} ms -> SLA Tier: {classify_sla_tier(lat)}")

Run:

uv run python sla_bisect_lookup.py

Output:

Latency:    4.2 ms -> SLA Tier: PLATINUM
Latency:   10.0 ms -> SLA Tier: GOLD
Latency:   25.0 ms -> SLA Tier: GOLD
Latency:   75.0 ms -> SLA Tier: SILVER
Latency:  450.0 ms -> SLA Tier: BRONZE
Latency: 2500.0 ms -> SLA Tier: BREACHED

Pitfalls

Pitfall 1: Modifying Elements in a heapq Directly

heapq functions (heappush, heappop) operate on a standard Python list. If you modify an element’s priority directly inside the list (queue[0] = new_val), you violate the binary heap invariant:

# THE BUG:
h = [1, 3, 5]
h[0] = 10  # Corrupts heap structure!
# heappop(h) now returns invalid data!

# THE FIX: Call heapq.heapify() to restore the heap invariant
heapq.heapify(h)

Exercises

  1. Use collections.defaultdict(set) to invert a graph mapping from node -> neighbors to neighbor -> incoming_nodes.
  2. Implement an execution worker pool that prioritizes urgent tasks using heapq.
  3. Use collections.deque(maxlen=100) to calculate a moving average over a simulated stream of 10,000 sensor numbers.
  4. Define a custom IntEnum for Linux file permissions (READ = 4, WRITE = 2, EXECUTE = 1) and demonstrate bitwise flags.

Further reading

  • Python Standard Library: collections, heapq, bisect, enum.
  • CPython Source: Modules/_collectionsmodule.c (C-optimized deque implementation).