The collections Module: Python's Supercharged Data Structures

# The full import (pick what you need)
from collections import (
    Counter,       # counting hashable objects
    defaultdict,   # dict with auto-created missing values
    OrderedDict,   # dict with order-sensitive operations
    namedtuple,    # immutable tuple with named fields
    deque,         # double-ended queue with O(1) both-end ops
    ChainMap,      # layered dict views
)
import heapq   # heap operations on lists
import bisect  # binary search on sorted lists

from collections import Counter

# Count characters in a string
char_count = Counter("banana")
print(char_count)
# Counter({'a': 3, 'n': 2, 'b': 1})

# Count words in a list
words = ["apple", "banana", "apple", "cherry", "banana", "apple"]
word_count = Counter(words)
print(word_count)
# Counter({'apple': 3, 'banana': 2, 'cherry': 1})

# Count words from a text
text = "to be or not to be that is the question to be"
token_counts = Counter(text.split())
print(token_counts)
# Counter({'to': 3, 'be': 3, 'or': 1, 'not': 1, ...})

from collections import Counter

corpus = [
    "the cat sat on the mat",
    "the dog ran in the park",
    "the cat and the dog played",
]

# Count all tokens across the corpus
all_tokens = []
for doc in corpus:
    all_tokens.extend(doc.split())

vocab_counter = Counter(all_tokens)

# Most common tokens (likely stopwords in real NLP)
print("Top 5 tokens:", vocab_counter.most_common(5))

# Build a vocabulary index (word -> id) from most common words
vocab_size = 10
vocab = {word: idx for idx, (word, _) in enumerate(vocab_counter.most_common(vocab_size))}
print("Vocab:", vocab)
# {'the': 0, 'cat': 1, 'dog': 2, ...}

# Encode a sentence
def encode(sentence: str, vocab: dict[str, int]) -> list[int]:
    """Convert words to vocab ids, unknown words get -1."""
    return [vocab.get(w, -1) for w in sentence.split()]

print(encode("the cat ran", vocab))   # [0, 1, ?]

from collections import defaultdict

# Without defaultdict -- classic verbose pattern
groups = {}
words = [("fruit", "apple"), ("veggie", "carrot"), ("fruit", "banana"), ("veggie", "broccoli")]
for category, item in words:
    if category not in groups:      # boilerplate check
        groups[category] = []       # boilerplate initialization
    groups[category].append(item)

print(groups)
# {'fruit': ['apple', 'banana'], 'veggie': ['carrot', 'broccoli']}


# With defaultdict(list) -- missing keys auto-get an empty list
groups_v2 = defaultdict(list)
for category, item in words:
    groups_v2[category].append(item)    # no check needed!

print(dict(groups_v2))
# {'fruit': ['apple', 'banana'], 'veggie': ['carrot', 'broccoli']}

from collections import OrderedDict

# OrderedDict has .move_to_end() which regular dicts lack
od = OrderedDict([("a", 1), ("b", 2), ("c", 3)])
od.move_to_end("a")          # move "a" to the end
print(list(od.keys()))       # ['b', 'c', 'a']

od.move_to_end("a", last=False)  # move "a" to the front
print(list(od.keys()))           # ['a', 'b', 'c']

# OrderedDict equality considers order; regular dict equality does not
d1 = OrderedDict([("a", 1), ("b", 2)])
d2 = OrderedDict([("b", 2), ("a", 1)])
print(d1 == d2)    # False -- order matters for OrderedDict

r1 = {"a": 1, "b": 2}
r2 = {"b": 2, "a": 1}
print(r1 == r2)    # True -- order does NOT matter for regular dict

from collections import OrderedDict

class LRUCache:
    """Least Recently Used cache using OrderedDict.

    The most recently used item moves to the end.
    When capacity is exceeded, the front (least recently used) is evicted.
    """

    def __init__(self, capacity: int) -> None:
        self._cache: OrderedDict = OrderedDict()
        self._capacity = capacity

    def get(self, key: str) -> object:
        if key not in self._cache:
            return -1
        self._cache.move_to_end(key)   # mark as recently used
        return self._cache[key]

    def put(self, key: str, value: object) -> None:
        if key in self._cache:
            self._cache.move_to_end(key)
        self._cache[key] = value
        if len(self._cache) > self._capacity:
            self._cache.popitem(last=False)  # evict least recently used (front)

    def __repr__(self) -> str:
        return f"LRUCache({dict(self._cache)})"


cache = LRUCache(capacity=3)
cache.put("a", 1)
cache.put("b", 2)
cache.put("c", 3)
print(cache.get("a"))   # 1 -- "a" is now most recently used
cache.put("d", 4)       # capacity exceeded, "b" is evicted (LRU)
print(cache.get("b"))   # -1 -- "b" was evicted
print(cache)            # LRUCache({'c': 3, 'a': 1, 'd': 4})

from collections import namedtuple

# Define a namedtuple type
Point = namedtuple("Point", ["x", "y"])
Color = namedtuple("Color", ["r", "g", "b"])
Employee = namedtuple("Employee", ["name", "department", "salary"])

# Create instances
p = Point(x=3, y=7)
c = Color(255, 128, 0)
emp = Employee("Alice", "Engineering", 95000)

# Named access (much clearer than p[0], p[1])
print(f"Point: ({p.x}, {p.y})")
print(f"Color: rgb({c.r}, {c.g}, {c.b})")
print(f"{emp.name} in {emp.department} earns ${emp.salary:,}")

# Still a tuple -- index access and unpacking work
print(p[0])                  # 3
x, y = p                     # unpacking
print(x, y)                  # 3, 7

# Immutable -- cannot modify fields
# p.x = 10    # AttributeError: can't set attribute

# _replace() creates a NEW namedtuple with one field changed
p2 = p._replace(x=10)
print(p2)   # Point(x=10, y=7)
print(p)    # Point(x=3, y=7) -- original unchanged

# _asdict() converts to a regular dict
print(emp._asdict())
# {'name': 'Alice', 'department': 'Engineering', 'salary': 95000}

from collections import deque
import time

# Demonstrate the performance gap
n = 100_000

# list.insert(0) is O(n) -- shifts every element
lst = []
start = time.time()
for i in range(n):
    lst.insert(0, i)
list_time = time.time() - start

# deque.appendleft() is O(1) -- no shifting
dq = deque()
start = time.time()
for i in range(n):
    dq.appendleft(i)
deque_time = time.time() - start

print(f"list.insert(0): {list_time:.3f}s")
print(f"deque.appendleft: {deque_time:.4f}s")
# list is typically 100-1000x slower for large n

from collections import deque

# maxlen: when full, adding to one end auto-discards from the other
history = deque(maxlen=5)
for i in range(10):
    history.append(i)
    print(f"Added {i}: {list(history)}")
# Added 0: [0]
# Added 1: [0, 1]
# ...
# Added 5: [1, 2, 3, 4, 5]  -- 0 was auto-discarded
# Added 9: [5, 6, 7, 8, 9]  -- always the last 5

# Sliding window average (useful for time-series smoothing)
def moving_average(data: list[float], window: int) -> list[float]:
    """Compute moving average using a deque window."""
    window_data: deque[float] = deque(maxlen=window)
    averages: list[float] = []

    for value in data:
        window_data.append(value)
        if len(window_data) == window:
            averages.append(sum(window_data) / window)

    return averages


prices = [10, 12, 11, 14, 15, 13, 16, 18, 17, 20]
ma = moving_average(prices, window=3)
print("Moving average (window=3):", [round(x, 2) for x in ma])
# [11.0, 12.33, 13.33, 14.0, 14.67, 15.67, 17.0, 18.33]

from collections import deque

def bfs_shortest_path(graph: dict, start: str, end: str) -> list[str] | None:
    """Find shortest path using BFS with deque."""
    if start == end:
        return [start]

    visited: set[str] = {start}
    # Each item in queue is a path so far
    queue: deque[list[str]] = deque([[start]])

    while queue:
        path = queue.popleft()
        current = path[-1]

        for neighbor in graph.get(current, []):
            if neighbor not in visited:
                new_path = path + [neighbor]
                if neighbor == end:
                    return new_path
                visited.add(neighbor)
                queue.append(new_path)

    return None   # no path found


graph = {
    "A": ["B", "C"],
    "B": ["A", "D", "E"],
    "C": ["A", "F"],
    "D": ["B"],
    "E": ["B", "F"],
    "F": ["C", "E"],
}

print(bfs_shortest_path(graph, "A", "F"))   # ['A', 'C', 'F']

from collections import ChainMap

# Classic use case: layered configuration
defaults = {
    "timeout": 30,
    "retries": 3,
    "debug": False,
    "log_level": "INFO",
}
env_vars = {
    "timeout": 60,
    "debug": True,
}
cli_args = {
    "log_level": "DEBUG",
}

# Priority: CLI > env vars > defaults
config = ChainMap(cli_args, env_vars, defaults)

print(config["timeout"])     # 60  (from env_vars)
print(config["retries"])     # 3   (from defaults)
print(config["log_level"])   # "DEBUG"  (from cli_args)
print(config["debug"])       # True  (from env_vars)

# Writes go to the first map only
config["new_key"] = "hello"
print(cli_args)   # {'log_level': 'DEBUG', 'new_key': 'hello'}
print(defaults)   # unchanged

# .maps gives access to the underlying list of dicts
print(config.maps)   # [cli_args, env_vars, defaults]

# Create a child context (adds a new layer at the front)
child = config.new_child({"timeout": 5})
print(child["timeout"])   # 5 (child override)
print(config["timeout"])  # 60 (parent unchanged)

import heapq

# heapq operates on a regular list
heap: list[int] = []
heapq.heappush(heap, 10)
heapq.heappush(heap, 3)
heapq.heappush(heap, 7)
heapq.heappush(heap, 1)
heapq.heappush(heap, 5)

print(heap)               # [1, 3, 7, 10, 5] (heap-ordered, not sorted)
print(heapq.heappop(heap))  # 1 (smallest always pops first)
print(heapq.heappop(heap))  # 3
print(heapq.heappop(heap))  # 5

# heapify: convert an existing list to a heap in O(n)
data = [8, 2, 6, 4, 1, 9, 3]
heapq.heapify(data)
print(data)               # [1, 2, 3, 4, 8, 9, 6] (heap property satisfied)

# nlargest / nsmallest: efficiently find top-K elements
scores = [45, 92, 78, 55, 88, 91, 63, 71, 84, 97]
print(heapq.nlargest(3, scores))   # [97, 92, 91]
print(heapq.nsmallest(3, scores))  # [45, 55, 63]

import heapq

max_heap: list[int] = []
for value in [5, 1, 8, 3, 7]:
    heapq.heappush(max_heap, -value)   # store negated

print(-heapq.heappop(max_heap))  # 8 (largest)
print(-heapq.heappop(max_heap))  # 7
print(-heapq.heappop(max_heap))  # 5

import heapq
from dataclasses import dataclass, field

@dataclass(order=True)
class Task:
    """A task with a priority and name."""
    priority: int
    name: str = field(compare=False)

task_queue: list[Task] = []
heapq.heappush(task_queue, Task(priority=3, name="Send report"))
heapq.heappush(task_queue, Task(priority=1, name="Fix critical bug"))
heapq.heappush(task_queue, Task(priority=2, name="Code review"))
heapq.heappush(task_queue, Task(priority=1, name="Deploy hotfix"))

# Process in priority order (lowest number = highest priority)
while task_queue:
    task = heapq.heappop(task_queue)
    print(f"  [{task.priority}] {task.name}")
# [1] Fix critical bug
# [1] Deploy hotfix
# [2] Code review
# [3] Send report

import heapq
import math

def k_closest(points: list[tuple[int, int]], k: int) -> list[tuple[int, int]]:
    """Return k points closest to the origin using a heap."""
    # Use a min-heap keyed by distance squared (avoid sqrt for speed)
    heap = [(p[0]**2 + p[1]**2, p) for p in points]
    heapq.heapify(heap)
    return [heapq.heappop(heap)[1] for _ in range(k)]


points = [(1, 3), (-2, 2), (5, 8), (0, 1), (3, -2)]
print(k_closest(points, k=2))   # [(0, 1), (-2, 2)]

import bisect

sorted_list = [1, 3, 5, 7, 9, 11, 15]

# bisect_left: index of leftmost position where x can be inserted
# (also the index where x would be if it already exists)
print(bisect.bisect_left(sorted_list, 7))    # 3 (7 is at index 3)
print(bisect.bisect_left(sorted_list, 6))    # 3 (6 would go at index 3)
print(bisect.bisect_left(sorted_list, 1))    # 0

# bisect_right (same as bisect.bisect): index AFTER any existing occurrences
sorted_with_dups = [1, 3, 3, 3, 5, 7]
print(bisect.bisect_left(sorted_with_dups, 3))    # 1 (before the 3s)
print(bisect.bisect_right(sorted_with_dups, 3))   # 4 (after the 3s)

# insort: insert while maintaining sorted order
data = [1, 3, 5, 7]
bisect.insort(data, 4)
print(data)   # [1, 3, 4, 5, 7]
bisect.insort(data, 6)
print(data)   # [1, 3, 4, 5, 6, 7]

import bisect

def grade(score: int) -> str:
    """Assign a letter grade using binary search."""
    breakpoints = [60, 70, 80, 90]
    grades      = ["F", "D", "C", "B", "A"]
    return grades[bisect.bisect(breakpoints, score)]


test_scores = [45, 62, 75, 83, 91, 100]
for s in test_scores:
    print(f"  {s} -> {grade(s)}")
# 45 -> F
# 62 -> D
# 75 -> C
# 83 -> B
# 91 -> A
# 100 -> A

import bisect

def count_in_range(sorted_data: list[int], lo: int, hi: int) -> int:
    """Count elements in [lo, hi] using binary search. O(log n)."""
    left  = bisect.bisect_left(sorted_data, lo)
    right = bisect.bisect_right(sorted_data, hi)
    return right - left

data = sorted([12, 5, 19, 3, 8, 15, 7, 22, 11, 4])
print(data)   # [3, 4, 5, 7, 8, 11, 12, 15, 19, 22]
print(count_in_range(data, 5, 15))   # 6 (values: 5, 7, 8, 11, 12, 15)
print(count_in_range(data, 10, 20))  # 4 (values: 11, 12, 15, 19)