itertools + functools: trusa de unelte de ordin superior

Există două module din stdlib care îți răsplătesc timpul investit în ele mai repede decât aproape orice altceva în Python: itertools și functools. Primul îți oferă algebră peste iteratori, adică să tai felii, să concatenezi, să grupezi, să combini, fără să rescrii la nesfârșit aceleași bucle imbricate. Al doilea îți oferă unelte la nivel de funcție, adică memorare în cache, currying, dispatch, lucruri normale în limbajele funcționale și ușor subutilizate aici.

Lecția asta e subsetul cu efect mare. Nu fiecare rețetă din ambele module, ci cele vreo duzină de membri pe care-i vei folosi iar și iar.

itertools: algebră peste iteratori

Tot ce e în itertools returnează un iterator, nu o listă. Asta înseamnă că e leneș, e ușor cu memoria, și trebuie să ții minte să-l înfășori în list(...) dacă vrei să te uiți la valori de mai multe ori.

chain: concatenează iterabile

from itertools import chain

a: list[int] = [1, 2, 3]
b: list[int] = [4, 5, 6]
c: tuple[int, ...] = (7, 8, 9)

list(chain(a, b, c))  # [1, 2, 3, 4, 5, 6, 7, 8, 9]

chain(*iterables) parcurge iterabilele în ordine, unul după altul. Avantajul mare față de a + b + c e că intrările nu trebuie să fie de același tip și nimic nu e materializat până nu iterezi. Există și chain.from_iterable(it) pentru cazul în care ai un iterabil de iterabile, aplatizând un singur nivel:

pages: list[list[int]] = [[1, 2], [3, 4], [5, 6]]
list(chain.from_iterable(pages))  # [1, 2, 3, 4, 5, 6]

islice: taie felii dintr-un iterabil

Nu poți face my_iterator[10:20] pentru că iteratorii nu suportă indexare. islice e substitutul conștient de iteratori:

from itertools import islice, count

# count() e infinit, islice ne salveaza
first_five: list[int] = list(islice(count(), 5))      # [0, 1, 2, 3, 4]
window: list[int] = list(islice(count(), 10, 15))     # [10, 11, 12, 13, 14]
every_other: list[int] = list(islice(count(), 0, 10, 2))  # [0, 2, 4, 6, 8]

Semnătura imită felierea obișnuită: islice(it, stop) sau islice(it, start, stop[, step]). Spre deosebire de felierea pe liste, indicii negativi nu sunt permiși, pentru că nu există un „capăt” al iteratorului de la care să numeri înapoi.

Cea mai obișnuită utilizare reală: paginarea peste un generator. Ai o funcție care produce rânduri de undeva costisitor și vrei pagina 3, câte 50 de elemente:

from collections.abc import Iterator
from itertools import islice

def fetch_rows() -> Iterator[dict]:
    ...

page_size: int = 50
page_num: int = 3
start: int = (page_num - 1) * page_size
page: list[dict] = list(islice(fetch_rows(), start, start + page_size))

tee: împarte în N iteratori independenți

from itertools import tee

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

list(a)  # [1, 2, 3, 4, 5]
list(b)  # [1, 2, 3, 4, 5]

tee(it, n) returnează n iteratori care produc fiecare aceleași valori ca originalul. Util când vrei să parcurgi aceeași secvență în două feluri fără s-o materializezi într-o listă, de exemplu calculând o sumă și un maxim într-o singură trecere fiecare.

Captura: tee funcționează tamponând valorile pe care un iterator le-a consumat și altul nu încă. Dacă o ramură o ia mult înaintea celeilalte, bufferul crește. Pentru consum echilibrat, e bine. Pentru „consumă tot a, apoi tot b”, ai materializat efectiv întreaga secvență în memorie, moment în care o listă ar fi fost mai simplă.

groupby: grupare secvențială

from itertools import groupby

data: list[tuple[str, int]] = [
    ("a", 1), ("a", 2), ("b", 3), ("b", 4), ("a", 5)
]

for key, group in groupby(data, key=lambda x: x[0]):
    print(key, list(group))
# a [('a', 1), ('a', 2)]
# b [('b', 3), ('b', 4)]
# a [('a', 5)]

Citește cu atenție. Cheia "a" apare de două ori în output, o dată la început, o dată la sfârșit. groupby grupează secvențe consecutive cu aceeași cheie. Nu sortează.

Dacă vrei semantica „group by key” în stil SQL, în care ordinea nu contează, ai două variante: ori sortezi întâi, ori folosești defaultdict(list) din lecția anterioară. groupby e unealta potrivită când datele sunt deja sortate sau când succesiunile consecutive sunt exact ce vrei: succesiuni ale aceluiași nivel de log, succesiuni ale aceluiași cod de status etc.

accumulate: totaluri parțiale

from itertools import accumulate
import operator

values: list[int] = [1, 2, 3, 4, 5]

list(accumulate(values))                   # [1, 3, 6, 10, 15] suma curenta
list(accumulate(values, operator.mul))     # [1, 2, 6, 24, 120] produs curent
list(accumulate(values, max))              # [1, 2, 3, 4, 5] maxim curent

accumulate(it, func=operator.add) produce rezultatul cumulat aplicând func de la stânga la dreapta. Implicit e adunarea, ceea ce-l face versiunea leneșă a „totalului curent”. Cu o funcție personalizată, orice are formă de fold funcționează.

Cuvântul cheie opțional initial (3.8+) îți permite să setezi o valoare de pornire:

list(accumulate([1, 2, 3], initial=100))  # [100, 101, 103, 106]

combinations, permutations, product: combinatorică

from itertools import combinations, permutations, product

list(combinations("abc", 2))   # [('a', 'b'), ('a', 'c'), ('b', 'c')]
list(permutations("abc", 2))   # [('a', 'b'), ('a', 'c'), ('b', 'a'), ('b', 'c'), ('c', 'a'), ('c', 'b')]
list(product([1, 2], ["x", "y"]))  # [(1, 'x'), (1, 'y'), (2, 'x'), (2, 'y')]

combinations(it, r): toate submulțimile de lungime r, ordinea nu contează, fără repetări.
permutations(it, r): toate ordonările de lungime r, ordinea contează, fără repetări.
product(*its, repeat=1): produs cartezian. Cea mai curată înlocuire pentru bucle for imbricate peste dimensiuni independente.

Înlocuirea buclelor imbricate:

# Inainte
for env in ["dev", "staging", "prod"]:
    for region in ["us", "eu", "ap"]:
        for size in ["small", "large"]:
            deploy(env, region, size)

# Dupa
for env, region, size in product(
    ["dev", "staging", "prod"],
    ["us", "eu", "ap"],
    ["small", "large"],
):
    deploy(env, region, size)

Același comportament, un singur nivel de indentare, mai ușor de adăugat o a patra dimensiune.

count, cycle, repeat: iteratori infiniți

from itertools import count, cycle, repeat

count(10)            # 10, 11, 12, 13, ... la nesfarsit
count(0, 0.25)       # 0, 0.25, 0.5, 0.75, ... la nesfarsit
cycle("ab")          # 'a', 'b', 'a', 'b', ... la nesfarsit
repeat("x", 3)       # 'x', 'x', 'x'
repeat("x")          # 'x', 'x', 'x', ... la nesfarsit

Asociază-i mereu cu ceva care se oprește: islice, zip cu un iterabil finit sau o condiție de break. Altfel, ai scris o buclă infinită.

O utilizare reală: numerotarea liniilor de log începând de la 1, indiferent câte sunt.

from itertools import count

for n, line in zip(count(1), open("log.txt"), strict=False):
    print(f"{n:>5}: {line}", end="")

zip cu iteratorul fișierului se oprește natural când fișierul se termină.

batched: împărțire în loturi (3.12+)

from itertools import batched

rows: list[int] = list(range(13))
for chunk in batched(rows, 5):
    print(chunk)
# (0, 1, 2, 3, 4)
# (5, 6, 7, 8, 9)
# (10, 11, 12)

batched(it, n) a fost adăugat în Python 3.12. Înainte, fiecare codebase Python avea propria funcție personalizată batched sau chunked. Acum nu mai ai nevoie de una. Util pentru gruparea apelurilor API, inserțiilor în baza de date, orice are o limită per cerere.

functools: unelte pentru funcții

reduce: da, e tot în regulă

from functools import reduce
import operator

reduce(operator.add, [1, 2, 3, 4, 5])         # 15
reduce(operator.mul, [1, 2, 3, 4, 5])         # 120
reduce(lambda a, b: a | b, [{1}, {2}, {3}])   # {1, 2, 3}

reduce(func, iterable[, initial]) pliază iterabilul de la stânga la dreapta cu funcția binară. Este versiunea în limbaj imperativ a fold-ului la stânga.

Guido a scos reduce din builtins în Python 3 pentru că, în vorbele lui, majoritatea utilizărilor sunt mai clare ca buclă. Are dreptate de cele mai multe ori, sum, min, max și any/all acoperă cazurile comune. Dar pentru fold-uri legitime care nu sunt unul dintre acelea, reuniuni de mulțimi, fuziuni de dicționare, monoizi personalizați, reduce e exact unealta potrivită. Nu-l evita din principiu.

# Fuzioneaza o lista de dicte (cheile de mai tarziu castiga)
from functools import reduce
configs: list[dict[str, str]] = [base, env_overrides, cli_overrides]
final: dict[str, str] = reduce(lambda a, b: {**a, **b}, configs, {})

partial: pre-completează argumente

from functools import partial

def power(base: float, exp: float) -> float:
    return base ** exp

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

square(5)  # 25
cube(5)    # 125

partial(func, *args, **kwargs) returnează un nou callable cu unele argumente fixate. Utilizări obișnuite:

Adaptarea unei funcții la o semnătură de callback care primește mai puține argumente.
Construirea de variante specializate fără să scrii funcții wrapper.
Pre-configurarea unui logger, unei sesiuni HTTP, unei conexiuni la baza de date.

import requests
from functools import partial

api_get = partial(requests.get, timeout=10, headers={"User-Agent": "myapp/1.0"})
api_get("https://example.com/users")  # timeout si headers deja aplicate

cache și lru_cache: memoizare

from functools import cache, lru_cache

@cache
def fib(n: int) -> int:
    if n < 2:
        return n
    return fib(n - 1) + fib(n - 2)

@lru_cache(maxsize=128)
def fetch_user(user_id: int) -> dict:
    ...  # apel costisitor

@cache (3.9+) e un cache nelimitat, fiecare apel cu argumente noi e memorat pentru totdeauna. @lru_cache(maxsize=N) e versiunea mărginită: păstrează cele N rezultate folosite cel mai recent, evacuează restul.

Folosește @cache nelimitat când:

Spațiul argumentelor e mic și cunoscut (numere întregi mici, string-uri fixe).
Memoizezi o funcție pură recursivă.

Folosește @lru_cache mărginit când:

Spațiul argumentelor poate crește la nesfârșit (ID-uri de utilizator, URL-uri, căi de fișiere).
Ai nevoie de un plafon de memorie.

Ambele cer ca argumentele să fie hashable. Nu decora o funcție care primește un list sau un dict ca argument, va crăpa la primul apel.

singledispatch: supraîncărcare de funcții după tip

from functools import singledispatch
from pathlib import Path
import json
import tomllib

@singledispatch
def load_config(source) -> dict:
    raise TypeError(f"Cannot load config from {type(source).__name__}")

@load_config.register
def _(source: Path) -> dict:
    text: str = source.read_text()
    if source.suffix == ".json":
        return json.loads(text)
    if source.suffix == ".toml":
        return tomllib.loads(text)
    raise ValueError(f"Unsupported file type: {source.suffix}")

@load_config.register
def _(source: dict) -> dict:
    return source

@load_config.register
def _(source: str) -> dict:
    return json.loads(source)

@singledispatch îți permite să scrii o funcție care face dispatch după tipul primului argument. Funcția de bază e fallback-ul. Fiecare @func.register adaugă o implementare pentru un tip specific, identificat prin adnotarea de tip de pe parametru.

Apelul load_config(Path("app.toml")) rulează ramura Path. Apelul load_config({"key": "value"}) rulează ramura dict. load_config(42) ridică TypeError din funcția de bază.

Acesta e răspunsul curat al lui Python la „vreau să tratez tipuri de input diferite în mod diferit fără un lanț de verificări isinstance”. E puțin mai greu decât un statement match, dar se compune: cod terț poate înregistra un handler pentru tipul propriu fără să atingă codul tău.

Notă despre asistența AI. Asistenții de cod sunt buni la a recunoaște când un lanț if isinstance(...) elif isinstance(...) elif ... ar putea deveni un @singledispatch. Sunt și buni la a-l folosi excesiv. Dacă ai două ramuri, lanțul e în regulă. Dispatch-ul își merită prețul la trei sau mai multe, mai ales când funcția ar putea fi extinsă mai târziu.

wraps: păstrează decoratorul cinstit

Acoperit deja în lecția 5. O reamintire rapidă: @functools.wraps(fn) pe funcția interioară a unui decorator copiază __name__, __doc__ și alte câteva atribute de la funcția înfășurată la wrapper. Fără el, fiecare funcție decorată arată ca wrapper pentru debuggere, generatoare de documentație și help().

Când stdlib nu e de ajuns: more-itertools

Pachetul PyPI more-itertools este extensia pentru producție. Lucruri ca chunked, windowed, unique_everseen, partition, take, interleave, rețete care au trăit ani buni în documentația oficială itertools ca exemple de tipul „ai putea scrie asta singur”, împachetate ca să nu fii nevoit.

from more_itertools import unique_everseen, windowed, partition

list(unique_everseen([1, 2, 1, 3, 2, 4]))  # [1, 2, 3, 4] pastreaza ordinea primei aparitii
list(windowed(range(5), 3))                # [(0,1,2), (1,2,3), (2,3,4)]
evens, odds = partition(lambda n: n % 2, range(10))

Dacă un coleg sau un asistent AI sugerează scrierea unui helper personalizat care sună ca un itertools, verifică întâi more-itertools.

Punând lucrurile cap la cap

Un mic exemplu care folosește bucăți din ambele module: prelucrarea unui flux de evenimente în rezumate orare.

from itertools import groupby
from functools import reduce
from collections import Counter
from datetime import datetime
from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class Event:
    timestamp: datetime
    user_id: int
    action: str

def hour_key(e: Event) -> datetime:
    return e.timestamp.replace(minute=0, second=0, microsecond=0)

def summarize(events: list[Event]) -> dict[datetime, Counter[str]]:
    events_sorted: list[Event] = sorted(events, key=lambda e: e.timestamp)
    grouped = groupby(events_sorted, key=hour_key)
    return {
        hour: reduce(lambda c, e: c + Counter([e.action]), group, Counter())
        for hour, group in grouped
    }

sorted ca să facem datele să se potrivească cu contractul de succesiuni consecutive al lui groupby. groupby ca să grupăm pe oră. reduce ca să pliem evenimentele fiecărei ore într-un Counter. Trei module, zece linii, fără condiționale imbricate.

Asta e trusa de unelte de ordin superior. Folosește-o când face intenția mai clară; întinde-te după o buclă simplă când nu o face.

Referințe: itertools — Functions creating iterators, functools — Higher-order functions, more-itertools, PEP 443 — Single-dispatch generic functions. Consultat 2026-05-01.