python -c "import itertools, functools; print(itertools.__name__, functools.__name__)"

itertools functools

import itertools as it
import functools as ft

from itertools import chain, groupby, islice, product, combinations
from functools import lru_cache, cache, partial, reduce, wraps, singledispatch, cached_property

from itertools import chain

list(chain([1, 2], [3, 4], [5]))
list(chain.from_iterable([[1, 2], [3, 4], [5]]))

[1, 2, 3, 4, 5]
[1, 2, 3, 4, 5]

from itertools import chain

rows = [{'tags': ['a', 'b']}, {'tags': ['c']}, {'tags': []}]
all_tags = list(chain.from_iterable(r['tags'] for r in rows))
print(all_tags)

['a', 'b', 'c']

from itertools import groupby

data = [
    {'dept': 'eng', 'name': 'Alice'},
    {'dept': 'eng', 'name': 'Bob'},
    {'dept': 'sales', 'name': 'Carol'},
    {'dept': 'eng', 'name': 'Dan'},
]

for key, group in groupby(data, key=lambda r: r['dept']):
    names = [r['name'] for r in group]
    print(key, names)

eng ['Alice', 'Bob']
sales ['Carol']
eng ['Dan']

from itertools import groupby
data_sorted = sorted(data, key=lambda r: r['dept'])
for key, group in groupby(data_sorted, key=lambda r: r['dept']):
    print(key, [r['name'] for r in group])

eng ['Alice', 'Bob', 'Dan']
sales ['Carol']

from itertools import accumulate
import operator

list(accumulate([1, 2, 3, 4, 5]))                       # running sum
list(accumulate([1, 2, 3, 4, 5], operator.mul))         # running product
list(accumulate([3, 1, 4, 1, 5, 9, 2, 6], max))         # running max
list(accumulate([1, 2, 3], initial=100))                # 3.8+

[1, 3, 6, 10, 15]
[1, 2, 6, 24, 120]
[3, 3, 4, 4, 5, 9, 9, 9]
[100, 101, 103, 106]

from itertools import pairwise, batched

list(pairwise('ABCDE'))                        # sliding window of 2
list(batched('ABCDEFG', 3))                    # chunks of 3 — last may be short

[('A', 'B'), ('B', 'C'), ('C', 'D'), ('D', 'E')]
[('A', 'B', 'C'), ('D', 'E', 'F'), ('G',)]

from itertools import pairwise
prices = [100, 102, 99, 104, 110]
deltas = [b - a for a, b in pairwise(prices)]
print(deltas)

[2, -3, 5, 6]

from itertools import islice, count

list(islice(count(), 5))                  # first 5 of an infinite counter
list(islice(count(), 10, 15))             # items 10..14
list(islice(count(), 0, 20, 4))           # every 4th item up to 20

[0, 1, 2, 3, 4]
[10, 11, 12, 13, 14]
[0, 4, 8, 12, 16]

from itertools import islice
with open('/etc/services') as f:
    for line in islice(f, 10):
        print(line.rstrip())

# /etc/services
#
# Network services, Internet style
...

from itertools import product, permutations, combinations

list(product('AB', repeat=2))            # cartesian — 4 results
list(permutations('ABC', 2))             # ordered — 6 results
list(combinations('ABC', 2))             # unordered — 3 results

[('A', 'A'), ('A', 'B'), ('B', 'A'), ('B', 'B')]
[('A', 'B'), ('A', 'C'), ('B', 'A'), ('B', 'C'), ('C', 'A'), ('C', 'B')]
[('A', 'B'), ('A', 'C'), ('B', 'C')]

from itertools import product
for combo in product(range(10), repeat=4):
    pin = ''.join(map(str, combo))
    if pin == '1337':
        print('found:', pin)
        break

found: 1337

from itertools import tee, count, cycle, repeat, islice

a, b = tee(iter([1, 2, 3, 4]))
list(a), list(b)

list(islice(count(100, 5), 5))                # 100, 105, 110, 115, 120
list(islice(cycle('AB'), 5))                  # 5 items from infinite cycle
list(repeat('x', 3))                          # finite when n given

([1, 2, 3, 4], [1, 2, 3, 4])
[100, 105, 110, 115, 120]
['A', 'B', 'A', 'B', 'A']
['x', 'x', 'x']

from itertools import takewhile, dropwhile

nums = [1, 4, 6, 4, 1, 9, 2]
list(takewhile(lambda x: x < 5, nums))
list(dropwhile(lambda x: x < 5, nums))

[1, 4]
[6, 4, 1, 9, 2]

from itertools import dropwhile
lines = ['# header', '# more header', 'data1', 'data2']
for line in dropwhile(lambda l: l.startswith('#'), lines):
    print(line)

data1
data2

from itertools import starmap, zip_longest, compress, filterfalse

list(starmap(pow, [(2, 3), (3, 2), (10, 0)]))
list(zip_longest('AB', 'xyz', fillvalue='-'))
list(compress('ABCDE', [1, 0, 1, 0, 1]))
list(filterfalse(lambda x: x % 2, range(8)))            # keep evens

[8, 9, 1]
[('A', 'x'), ('B', 'y'), ('-', 'z')]
['A', 'C', 'E']
[0, 2, 4, 6]

from functools import lru_cache, cache

@lru_cache(maxsize=1024)
def fib(n):
    return n if n < 2 else fib(n - 1) + fib(n - 2)

print(fib(100))
print(fib.cache_info())
fib.cache_clear()

354224848179261915075
CacheInfo(hits=98, misses=101, maxsize=1024, currsize=101)

from functools import cache

@cache
def is_prime(n):
    if n < 2:
        return False
    return all(n % i for i in range(2, int(n ** 0.5) + 1))

print([n for n in range(30) if is_prime(n)])

[2, 3, 5, 7, 11, 13, 17, 19, 23, 29]

from functools import cached_property
import time

class Report:
    def __init__(self, rows):
        self.rows = rows

    @cached_property
    def total(self):
        time.sleep(0.5)             # pretend this is expensive
        return sum(r['amount'] for r in self.rows)

r = Report([{'amount': 10}, {'amount': 20}, {'amount': 30}])
print(r.total)                       # slow — computes
print(r.total)                       # instant — cached

60
60

from functools import partial

def power(base, exp):
    return base ** exp

square = partial(power, exp=2)
cube = partial(power, exp=3)

print(square(7))
print(cube(3))
print(list(map(partial(int, base=16), ['ff', '0a', '10'])))

49
27
[255, 10, 16]

from functools import partial
def on_click(button_id, event):
    print(f'button {button_id} fired {event}')

handler = partial(on_click, 'save')
handler('mouseup')

button save fired mouseup

from functools import reduce
import operator

reduce(operator.add, [1, 2, 3, 4, 5])              # same as sum()
reduce(operator.mul, [1, 2, 3, 4, 5], 1)           # product
reduce(lambda a, b: {**a, **b}, [{'a': 1}, {'b': 2}, {'c': 3}], {})

15
120
{'a': 1, 'b': 2, 'c': 3}

from functools import wraps
import time

def timed(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        t0 = time.perf_counter()
        result = func(*args, **kwargs)
        print(f'{func.__name__} took {time.perf_counter() - t0:.4f}s')
        return result
    return wrapper

@timed
def slow(n):
    """Pretend to be slow."""
    time.sleep(n)

slow(0.1)
print(slow.__name__, '—', slow.__doc__)

slow took 0.1003s
slow — Pretend to be slow.

from functools import singledispatch
from pathlib import Path

@singledispatch
def describe(arg):
    return f'object of type {type(arg).__name__}'

@describe.register
def _(arg: int):
    return f'int {arg}'

@describe.register
def _(arg: list):
    return f'list of {len(arg)}'

@describe.register(Path)
def _(arg):
    return f'path {arg}'

print(describe(42))
print(describe([1, 2, 3]))
print(describe(Path('/tmp/x')))
print(describe(3.14))

int 42
list of 3
path /tmp/x
object of type float

from functools import total_ordering

@total_ordering
class Version:
    def __init__(self, s):
        self.parts = tuple(int(p) for p in s.split('.'))

    def __eq__(self, other):
        return self.parts == other.parts

    def __lt__(self, other):
        return self.parts < other.parts

a = Version('1.2.3')
b = Version('1.2.10')
print(a < b, a <= b, a > b, a >= b, a == b)

True True False False False

from itertools import islice
from functools import lru_cache

def windows(seq, n):
    it = iter(seq)
    window = tuple(islice(it, n))
    if len(window) == n:
        yield window
    for x in it:
        window = window[1:] + (x,)
        yield window

@lru_cache(maxsize=None)
def price_for(day):
    return {1: 10, 2: 12, 3: 11, 4: 14, 5: 13}[day]

days = list(range(1, 6))
prices = [price_for(d) for d in days]
avgs = [round(sum(w) / 3, 2) for w in windows(prices, 3)]
print(avgs)

[11.0, 12.33, 12.67]

from itertools import batched
from functools import lru_cache

@lru_cache(maxsize=10_000)
def seen(item):
    return item

def paginate_dedup(stream, n):
    out = []
    for batch in batched(stream, n):
        page = []
        for item in batch:
            if seen.cache_info().currsize == 0 or item not in seen.__wrapped__.__globals__:
                seen(item)                  # cache it
                page.append(item)
        if page:
            out.append(page)
    return out

stream = ['a', 'b', 'a', 'c', 'd', 'b', 'e', 'f', 'a']
print(paginate_dedup(stream, 3))

[['a', 'b'], ['c', 'd'], ['e', 'f']]

from itertools import groupby
from functools import reduce

words = sorted('the quick brown fox jumps over the lazy fox'.split())
counts = {k: reduce(lambda acc, _: acc + 1, g, 0) for k, g in groupby(words)}
print(counts)

{'brown': 1, 'fox': 2, 'jumps': 1, 'lazy': 1, 'over': 1, 'quick': 1, 'the': 2}

from functools import partial, reduce

def add(x, y): return x + y
def mul(x, y): return x * y

pipeline = [partial(add, 10), partial(mul, 2), partial(add, -5)]

def compose(value, fn):
    return fn(value)

print(reduce(compose, pipeline, 3))           # ((3 + 10) * 2) - 5 = 21

21

import csv
from itertools import groupby
from io import StringIO

data = StringIO("""category,amount
food,10
food,20
travel,50
travel,75
food,5
""")

reader = csv.DictReader(data)
for cat, rows in groupby(reader, key=lambda r: r['category']):
    total = sum(int(r['amount']) for r in rows)
    print(f'{cat:8s} {total}')

food     30
travel   125
food     5

import time
from functools import wraps, partial

def retry(func=None, *, times=3, delay=0.1):
    if func is None:
        return partial(retry, times=times, delay=delay)

    @wraps(func)
    def wrapper(*args, **kwargs):
        last_exc = None
        for _ in range(times):
            try:
                return func(*args, **kwargs)
            except Exception as exc:
                last_exc = exc
                time.sleep(delay)
        raise last_exc
    return wrapper

@retry(times=3, delay=0.01)
def flaky(n):
    """Fail twice, succeed third time."""
    flaky.calls += 1
    if flaky.calls < 3:
        raise RuntimeError('not yet')
    return n * 2

flaky.calls = 0
print(flaky(21))
print(flaky.__name__, '—', flaky.__doc__)

42
flaky — Fail twice, succeed third time.

from functools import cache

@cache
def parse_expr(tokens, pos):
    if pos >= len(tokens):
        return None, pos
    if tokens[pos].isdigit():
        return int(tokens[pos]), pos + 1
    return None, pos

tokens = ('1', '+', '2', '*', '3')
print(parse_expr(tokens, 0))
print(parse_expr(tokens, 2))
print(parse_expr.cache_info())

(1, 1)
(2, 3)
CacheInfo(hits=0, misses=2, maxsize=None, currsize=2)

from itertools import product

grid = {
    'lr':       [0.1, 0.01, 0.001],
    'momentum': [0.0, 0.9],
    'batch':    [32, 64],
}

best = (None, -1)
for lr, mom, bs in product(*grid.values()):
    score = (1 / lr) * (1 + mom) - (bs * 0.001)
    if score > best[1]:
        best = ((lr, mom, bs), score)
print('best:', best)

best: ((0.001, 0.9, 32), 1899.968)

from functools import singledispatch
from datetime import datetime, date
from pathlib import Path
from decimal import Decimal
import json

@singledispatch
def to_json(value):
    raise TypeError(f'cannot encode {type(value).__name__}')

@to_json.register
def _(v: datetime): return v.isoformat()
@to_json.register
def _(v: date):     return v.isoformat()
@to_json.register
def _(v: Path):     return str(v)
@to_json.register
def _(v: Decimal):  return float(v)

payload = {
    'when':   datetime(2026, 5, 25, 12, 0),
    'where':  Path('/var/log/app.log'),
    'amount': Decimal('19.95'),
}
print(json.dumps(payload, default=to_json))

{"when": "2026-05-25T12:00:00", "where": "/var/log/app.log", "amount": 19.95}

import sys
from itertools import groupby

argv = ['--name', 'alice', '--tag', 'a', '--tag', 'b', 'file1', 'file2']

flags, positional = [], []
i = 0
while i < len(argv):
    if argv[i].startswith('--'):
        flags.append((argv[i][2:], argv[i + 1]))
        i += 2
    else:
        positional.append(argv[i])
        i += 1

grouped = {k: [v for _, v in g] for k, g in groupby(sorted(flags), key=lambda x: x[0])}
print(grouped)
print(positional)

{'name': ['alice'], 'tag': ['a', 'b']}
['file1', 'file2']