High-Performance Iteration and Functional Tools

Updated

September 7, 2026

High-Performance Iteration and Functional Tools

After reading this chapter, you will master C-accelerated iterator pipelines with itertools (chain, batched, groupby, islice), implement high-speed memoization with functools.lru_cache, construct pre-configured callables with functools.partial, and implement clean polymorphic function overloading with functools.singledispatch.

Mental model

Python loops written in pure Python incur bytecode interpreter evaluation overhead on every iteration. The itertools module executes inside compiled C routines (itertoolsmodule.c), pulling items directly across memory without constructing intermediate Python lists:

Pure Python List Intermediate (RAM Heavy & Slow):
  data ──▶ [f(x) for x in data] ──▶ [g(x) for x in ...] ──▶ Result List

Itertools C-Streaming Pipeline (Zero Intermediate Memory & High Speed):
  data ──▶ islice() ──▶ chain() ──▶ batched() ──▶ Consumer
             ▲           ▲            ▲
             └───────────┴────────────┴── Direct C-level iterator chaining

Minimal example

Save as itertools_functools_showcase.py:

# itertools_functools_showcase.py
import functools
import itertools
import time

# 1. functools.lru_cache: High-performance memoization
@functools.lru_cache(maxsize=128)
def resolve_ip_geo(ip_address: str) -> str:
    """Simulate expensive external geo-IP database lookup."""
    time.sleep(0.01)  # Simulated latency
    return "US-EAST" if ip_address.startswith("10.") else "EU-WEST"

def main() -> None:
    # 2. itertools.batched (Python 3.12+): Efficient batch chunking
    records = [f"rec_{i:02d}" for i in range(11)]
    print("Batching 11 records into chunks of 3:")
    for batch in itertools.batched(records, 3):
        print(f"  Processed batch: {batch}")

    # 3. itertools.chain: Flattening multiple streams seamlessly
    stream_a = ["event_1", "event_2"]
    stream_b = ["event_3", "event_4"]
    combined = list(itertools.chain(stream_a, stream_b))
    print(f"\nChained streams: {combined}")

    # Cache testing
    print("\nTesting LRU Cache performance:")
    t0 = time.perf_counter()
    res1 = resolve_ip_geo("10.0.1.5")  # Cache miss (sleeps 10ms)
    miss_duration = time.perf_counter() - t0

    t0 = time.perf_counter()
    res2 = resolve_ip_geo("10.0.1.5")  # Cache hit (instant)
    hit_duration = time.perf_counter() - t0

    print(f"First call (Miss): {miss_duration*1000:.2f} ms")
    print(f"Second call (Hit): {hit_duration*1000:.2f} ms")
    print(f"Cache stats      : {resolve_ip_geo.cache_info()}")

if __name__ == "__main__":
    main()

Run via uv run python itertools_functools_showcase.py:

Batching 11 records into chunks of 3:
  Processed batch: ('rec_00', 'rec_01', 'rec_02')
  Processed batch: ('rec_03', 'rec_04', 'rec_05')
  Processed batch: ('rec_06', 'rec_07', 'rec_08')
  Processed batch: ('rec_09', 'rec_10')

Chained streams: ['event_1', 'event_2', 'event_3', 'event_4']

Testing LRU Cache performance:
First call (Miss): 10.15 ms
Second call (Hit): 0.00 ms
Cache stats      : CacheInfo(hits=1, misses=1, maxsize=128, currsize=1)

Worked examples

Case 1: Telemetry Grouping with itertools.groupby()

itertools.groupby() groups adjacent elements sharing a common key. Critical Rule: It only groups consecutive elements, so inputs must be sorted by the key attribute first:

# telemetry_groupby.py
import itertools
from operator import itemgetter

def group_metrics_by_region(events: list[dict[str, str | float]]) -> None:
    # 1. MUST sort by group key first
    events.sort(key=itemgetter("region"))

    # 2. Group adjacent records
    print("Grouped telemetry metrics:")
    for region, items_iter in itertools.groupby(events, key=itemgetter("region")):
        items = list(items_iter)
        avg_latency = sum(float(x["latency"]) for x in items) / len(items)
        print(f"Region: {region:10} | Nodes: {len(items)} | Avg Latency: {avg_latency:.1f}ms")

if __name__ == "__main__":
    raw_events = [
        {"region": "us-east", "node": "n1", "latency": 12.0},
        {"region": "eu-west", "node": "n2", "latency": 88.0},
        {"region": "us-east", "node": "n3", "latency": 16.0},
        {"region": "eu-west", "node": "n4", "latency": 92.0},
    ]
    group_metrics_by_region(raw_events)

Run:

uv run python telemetry_groupby.py

Output:

Grouped telemetry metrics:
Region: eu-west    | Nodes: 2 | Avg Latency: 90.0ms
Region: us-east    | Nodes: 2 | Avg Latency: 14.0ms

Case 2: Clean Polymorphism with functools.singledispatch

Instead of writing fragile if isinstance(val, int): ... elif isinstance(val, dict): ... trees, @singledispatch decouples type handling into modular functions:

# event_serializer.py
from functools import singledispatch
from datetime import datetime, date

@singledispatch
def serialize(val: object) -> str:
    """Default fallback serializer."""
    return f"STR:{val}"

@serialize.register
def _(val: int | float) -> str:
    return f"NUM:{val}"

@serialize.register
def _(val: date) -> str:
    return f"DATE:{val.isoformat()}"

@serialize.register
def _(val: dict) -> str:
    items = [f"{k}={serialize(v)}" for k, v in val.items()]
    return f"MAP:{{{', '.join(items)}}}"

if __name__ == "__main__":
    print(serialize(42))
    print(serialize("active"))
    print(serialize(date(2026, 9, 7)))
    print(serialize({"id": 101, "name": "worker"}))

Run:

uv run python event_serializer.py

Output:

NUM:42
STR:active
DATE:2026-09-07
MAP:{id=NUM:101, name=STR:worker}

Pitfalls

Pitfall 1: Unsorted Input to itertools.groupby

If items with the same key are separated by other items, groupby() yields duplicate groups rather than consolidating them:

# THE BUG:
data = [("A", 1), ("B", 2), ("A", 3)]
for k, g in itertools.groupby(data, key=lambda x: x[0]):
    print(k, list(g))
# Output has two separate 'A' groups!
# A [('A', 1)]
# B [('B', 2)]
# A [('A', 3)]

Always call data.sort(key=...) before passing data to itertools.groupby().

Pitfall 2: Decorating Functions with Mutable Arguments in @lru_cache

lru_cache builds internal dictionary keys from function arguments. Passing a mutable type (list, dict, set) raises TypeError: unhashable type: 'list'. Always pass immutable arguments (tuples, strings, ints) into cached functions.

Exercises

  1. Use itertools.cycle to implement a round-robin DNS load balancer that cycles through three server IP addresses.
  2. Given a list of 100 integers, use itertools.islice() to inspect items 20 through 30 without loading the rest into memory.
  3. Use functools.partial to create a pre-configured log_error callable from print(..., file=sys.stderr, flush=True).
  4. Implement a combinatorial test case generator using itertools.product across 3 OS options and 4 Python versions.

Further reading

  • Python Standard Library: itertools and functools documentation.
  • Python Recipes: Itertools Recipes (Official documentation appendix).
  • PEP 443: Single-dispatch generic functions.