Идиоматичный Python: генераторы, итераторы, контекстные менеджеры, comprehensions

«Идиоматичный» — слово опасное: его легко свести к списку косметических правил вроде «не пиши for i in range(len(xs))». Настоящая идиоматика Python устроена иначе: язык почти не имеет встроенных конструкций, зато имеет протоколы — наборы dunder-методов, которые открывают доступ к синтаксису. Реализовал __iter__ — работает for, распаковка, in, sum, zip, sorted, *args. Реализовал __enter__/__exit__ — работает with со всеми гарантиями раскрутки стека. Идиоматичный код — это код, который говорит на языке протоколов, а не изобретает свои механизмы поверх них.

Практический выигрыш от этого двойной. Первый — читаемость: функция, принимающая Iterable[str], одинаково работает со списком, файлом, курсором БД и сетевым потоком. Второй — память и латентность: ленивый пайплайн обрабатывает поток любого размера в константной памяти и отдаёт первый результат до того, как прочитан последний байт входа.

Базовый синтаксис циклов и списков разбирался в курсе Программирование с нуля; модель ссылок — в модели данных; устройство list/dict — в коллекциях; замыкания и декораторы — в функциях и областях видимости. Здесь мы идём глубже: как именно работает for, во что превращается yield, почему @contextmanager без try/finally — это баг, и где ленивость честно проигрывает.

Карта протоколов

Всё, о чём пойдёт речь, — четыре связанных протокола плюс их асинхронные близнецы.

Все они формализованы в collections.abc и проверяются isinstance без наследования — у соответствующих ABC есть __subclasshook__. Это тот самый мостик между утиной типизацией и статической проверкой, который в типизации превращается в Protocol.

Протокол итерации: что на самом деле делает for

for x in obj — синтаксический сахар. Интерпретатор выполняет ровно три шага: получает итератор через iter(obj), в цикле дёргает next(it), и завершается, поймав StopIteration. Никакой «длины», «индексов» и «конца коллекции» в этом протоколе нет.

Эквивалент на чистом Python:

def manual_for(obj, body):
    """Ровно то, что делает for — без сахара."""
    it = iter(obj)                 # 1. получаем итератор
    while True:
        try:
            value = next(it)       # 2. тянем следующий элемент
        except StopIteration:      # 3. поток исчерпан — выходим
            break
        body(value)

manual_for([10, 20, 30], print)
# 10
# 20
# 30

В байткоде это GET_ITER + FOR_ITER; FOR_ITER умеет ловить StopIteration без раскрутки полноценного исключения, поэтому цикл дёшев.

Iterable ≠ Iterator

Различие критично, и его путают чаще всего. Iterable — то, из чего можно получить итератор (список, строка, dict). Iterator — одноразовый курсор с состоянием; по контракту его __iter__ возвращает self.

xs = [1, 2, 3]
print(iter(xs) is iter(xs))     # False — каждый раз новый курсор

it = iter(xs)
print(iter(it) is it)           # True — итератор возвращает сам себя

print(list(it))                 # [1, 2, 3]
print(list(it))                 # []  — исчерпан, второй проход пуст

Отсюда главная грабля ленивого кода: генератор, map, filter, zip, csv.reader и файловый объект — это итераторы, а не коллекции. Их нельзя обойти дважды, у них нет len(), их нельзя индексировать, а if x in it частично их съедает.

rows = (line.strip() for line in ["a", "b", "c"])
print(sum(1 for _ in rows))     # 3
print(list(rows))               # [] — данных больше нет

Если нужен повторный обход — либо материализуйте (rows = list(rows)), либо сделайте переиспользуемый iterable: класс с __iter__, который каждый раз создаёт новый генератор.

from collections.abc import Iterator
from pathlib import Path


class LogFile:
    """Iterable (не iterator): каждый обход открывает файл заново."""

    def __init__(self, path: Path) -> None:
        self._path = path

    def __iter__(self) -> Iterator[str]:
        with self._path.open(encoding="utf-8") as fh:
            for line in fh:
                yield line.rstrip("\n")


log = LogFile(Path("app.log"))
errors = sum(1 for line in log if "ERROR" in line)   # первый проход
warns = sum(1 for line in log if "WARN" in line)     # второй проход — работает

Цена честная: файл читается дважды. Зато семантика предсказуема, а память константна.

Свой итератор классом — и почему так почти никогда не пишут

class Countdown:
    """Классический итератор «руками»: пять строк состояния и два метода."""

    def __init__(self, start: int) -> None:
        self._n = start

    def __iter__(self) -> "Countdown":
        return self

    def __next__(self) -> int:
        if self._n <= 0:
            raise StopIteration
        self._n -= 1
        return self._n + 1


print(list(Countdown(3)))   # [3, 2, 1]

То же самое генератором:

def countdown(start: int):
    """Состояние хранит сам кадр функции — вручную его вести не нужно."""
    while start > 0:
        yield start
        start -= 1


print(list(countdown(3)))   # [3, 2, 1]

Генератор выигрывает не только в объёме. В классе состояние приходится расщеплять на поля и восстанавливать из них позицию в алгоритме; в генераторе состоянием является точка выполнения — то есть сам алгоритм. Для обхода дерева разница между двумя подходами — это разница между явным стеком и тремя строками рекурсии.

Двухаргументный iter — забытая идиома

iter(callable, sentinel) вызывает функцию без аргументов, пока та не вернёт сторожевое значение. Это превращает «цикл с while True и break» в обычный iterable.

from functools import partial

with open("dump.bin", "rb") as fh:
    # читаем блоками по 64 КиБ, пока read не вернёт b"" — конец файла
    for chunk in iter(partial(fh.read, 65536), b""):
        process(chunk)

Тот же приём: iter(input, ""), iter(queue.get, SENTINEL), iter(cursor.fetchone, None).

Расширения протокола: __reversed__, __contains__, __length_hint__

reversed(obj) требует либо __reversed__, либо пары __len__ + __getitem__; на генераторе он не работает — обратный порядок для потока не определён. in использует __contains__, а при его отсутствии линейно обходит итератор. __length_hint__ (PEP 424) — необязательная подсказка размера: её использует list(gen), чтобы преаллоцировать буфер.

Есть и легаси-путь: если у объекта нет __iter__, но есть __getitem__, iter() построит итератор, дёргающий индексы 0, 1, 2… до IndexError. Старый код на этом держится, новый писать так не стоит.

Comprehensions: выражение вместо цикла

Comprehension — это не «краткая запись цикла», а выражение, описывающее результат. Разница видна по тому, как код читается вслух: [t.name for t in tasks if t.is_active] читается как «имена активных задач», а цикл с append — как «создай пустой список, потом в него что-то добавляй».

tasks = [{"name": "build", "ok": True}, {"name": "lint", "ok": False}]

names = [t["name"] for t in tasks if t["ok"]]        # list
by_name = {t["name"]: t["ok"] for t in tasks}        # dict
failed = {t["name"] for t in tasks if not t["ok"]}   # set
lazy = (t["name"] for t in tasks)                    # генераторное выражение

print(names, by_name, failed)
# ['build'] {'build': True, 'lint': False} {'lint'}

Порядок for-ов во вложенных comprehension читается слева направо, как обычные вложенные циклы, а выражение результата стоит впереди:

matrix = [[1, 2], [3, 4]]
flat = [x for row in matrix for x in row]        # [1, 2, 3, 4]
pairs = [(i, j) for i in range(3) for j in range(i)]
print(flat, pairs)
# [1, 2, 3, 4] [(1, 0), (2, 0), (2, 1)]

Когда comprehension вреден

Три надёжных признака, что нужен обычный цикл:

1. Больше двух уровней вложенности или два if. Читаемость обрушивается быстрее, чем растёт краткость.
2. Побочные эффекты. [send(x) for x in xs] строит и выбрасывает список из None и вводит читателя в заблуждение. Нужен эффект — пишите цикл.
3. Обработка ошибок внутри. В comprehension нет try. Городить хелпер-функцию ради этого — хуже, чем написать четыре строки цикла.

# плохо: список создаётся ради побочного эффекта
[logger.info(row) for row in rows]

# хорошо
for row in rows:
    logger.info(row)

Скобки решают: список против генератора

import tracemalloc


def peak(fn, *args) -> float:
    tracemalloc.start()
    fn(*args)
    _, top = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return top / 1024 / 1024


eager = lambda n: sum([i * i for i in range(n)])   # сначала список на n элементов
lazy = lambda n: sum(i * i for i in range(n))      # поток, память константна

print(f"список:   {peak(eager, 1_000_000):.1f} МиБ")
print(f"генератор: {peak(lazy, 1_000_000):.1f} МиБ")
# список:   38.6 МиБ
# генератор: 0.0 МиБ

Числа приблизительные (CPython 3.12, x86-64), но порядок величины устойчив: список из миллиона целых — это 8 МиБ указателей плюс ~28–32 байта на каждый объект int.

Важное исключение, о котором почти не пишут: str.join внутри себя всё равно материализует последовательность, поэтому "".join([...]) быстрее, чем "".join(... for ...). А any/all наоборот — им генератор даёт короткое замыкание и они не читают вход дальше первого совпадения.

# join материализует аргумент — списочный comprehension здесь быстрее
csv_line = ",".join([str(v) for v in values])

# any коротко замыкается — генератор не прочитает весь файл
has_error = any("ERROR" in line for line in log_lines)

Стоимость: цифры, а не ощущения

Порядок величины на CPython 3.12 для миллиона элементов, timeit:

Вывод прагматичный: comprehension — разумный дефолт; map с готовой C-функцией выигрывает; map с лямбдой почти всегда проигрывает; генератор берут ради памяти и стриминга, а не ради скорости. В 3.12 list/dict/set-comprehension-ы инлайнятся в кадр вызывающей функции — отдельный кадр больше не создаётся; генераторные выражения по-прежнему компилируются в отдельную функцию, потому что им нужен собственный приостанавливаемый кадр.

Моржовый оператор в comprehension

:= (PEP 572) избавляет от двойного вычисления дорогой функции:

# без walrus: parse() вызывается дважды на каждый элемент
good = [parse(x) for x in raw if parse(x) is not None]

# с walrus: один вызов, результат переиспользуется
good = [p for x in raw if (p := parse(x)) is not None]

Генераторы: функция, которую можно поставить на паузу

Функция с yield в теле не выполняется при вызове. Вызов создаёт объект-генератор, внутри которого лежит замороженный кадр: локальные переменные, указатель инструкции, стек значений. next() размораживает кадр, доводит до ближайшего yield, забирает значение и снова замораживает.

import inspect


def gen():
    print("старт")
    x = yield 1
    print(f"получено: {x}")
    yield 2
    print("финал")


g = gen()
print(inspect.getgeneratorstate(g))   # GEN_CREATED — тело ещё не выполнялось
print(next(g))                        # старт / 1
print(inspect.getgeneratorstate(g))   # GEN_SUSPENDED
print(g.send("привет"))               # получено: привет / 2
try:
    next(g)
except StopIteration:
    print(inspect.getgeneratorstate(g))   # финал / GEN_CLOSED

Полный интерфейс: send, throw, close, return

yield — не только «отдать значение», но и точка входа обратно в генератор.

• gen.send(v) — возобновляет генератор, и выражение yield возвращает v. Первый send обязан быть send(None) (или next), иначе TypeError: кадр ещё не дошёл до yield.
• gen.throw(exc) — бросает исключение в точке yield, где его можно поймать.
• gen.close() — бросает GeneratorExit. Генератор обязан завершиться; если он попробует сделать ещё один yield, получите RuntimeError: generator ignored GeneratorExit.
• return value внутри генератора кладёт значение в StopIteration.value — для for оно невидимо, но его забирает yield from.

def accumulator():
    """Сопрограмма-накопитель: тянет значения через send, отдаёт текущую сумму."""
    total = 0
    try:
        while True:
            x = yield total          # отдаём сумму, ждём следующее слагаемое
            total += x
    except GeneratorExit:
        print(f"закрыт, итог {total}")   # гарантированная финализация


acc = accumulator()
next(acc)                  # праймим: доходим до первого yield
print(acc.send(10))        # 10
print(acc.send(5))         # 15
acc.close()                # закрыт, итог 15

Это исторический предшественник async/await (PEP 342, PEP 492). Сегодня писать сопрограммы на send не нужно — есть asyncio, см. конкурентность. Но понимать механику стоит: именно так под капотом работает await.

yield from: делегирование, а не сахар для цикла

def flatten(node):
    """Обход вложенных списков любой глубины."""
    for item in node:
        if isinstance(item, list):
            yield from flatten(item)     # делегируем подгенератору
        else:
            yield item


print(list(flatten([1, [2, [3, [4]], 5], 6])))   # [1, 2, 3, 4, 5, 6]

yield from sub эквивалентен циклу for v in sub: yield v только по значениям. Дополнительно он прокидывает send/throw/close в подгенератор и подставляет результат return подгенератора как значение выражения (PEP 380):

def reader():
    total = 0
    while (chunk := (yield)) is not None:
        total += len(chunk)
    return total                       # уйдёт в StopIteration.value


def wrapper():
    size = yield from reader()          # ловим return подгенератора
    print(f"прочитано {size} байт")

Ограничение честное: делегирование не бесплатно. Стоимость одного элемента линейна по глубине цепочки yield from, поэтому рекурсивный обход дерева глубиной 1000 на генераторах будет заметно медленнее явного стека.

Грабля: StopIteration внутри генератора

Классическая ловушка, закрытая PEP 479 (по умолчанию с 3.7): исключение StopIteration, случайно вылетевшее из тела генератора, раньше молча обрывало цикл. Теперь оно конвертируется в RuntimeError.

def take_until_blank(lines):
    it = iter(lines)
    while True:
        line = next(it)        # когда строки кончатся — StopIteration изнутри тела
        if not line:
            return
        yield line


print(list(take_until_blank(["a", "b"])))
# RuntimeError: generator raised StopIteration

Правильно: next(it, None) с дефолтом либо явный try/except StopIteration: return.

Грабля: with внутри брошенного генератора

Если генератор с открытым файлом не дочитали и не закрыли, finally/__exit__ выполнятся только при сборке мусора. На CPython счётчик ссылок обычно спасает сразу, но на PyPy, при циклических ссылках или когда генератор попал в долгоживущую структуру, дескриптор останется открытым.

from contextlib import closing

def head(gen, n):
    """Берём первые n элементов и ЯВНО закрываем источник."""
    with closing(gen) as g:          # гарантированный g.close() → GeneratorExit → finally
        for i, item in enumerate(g):
            if i >= n:
                break
            yield item

Для асинхронных генераторов аналог — contextlib.aclosing (3.10+), и там это уже не «хорошая практика», а необходимость: событийный цикл может завершиться раньше, чем сборщик доберётся до генератора.

Ленивые пайплайны

Собранная из генераторов цепочка ведёт себя как конвейер с «вытягиванием»: потребитель дёргает next, запрос идёт вверх по цепочке, обратно спускается ровно один элемент.

from collections.abc import Iterable, Iterator
from pathlib import Path


def read_lines(path: Path) -> Iterator[str]:
    with path.open(encoding="utf-8") as fh:
        yield from fh


def strip_and_skip(lines: Iterable[str]) -> Iterator[str]:
    for line in lines:
        line = line.strip()
        if line and not line.startswith("#"):
            yield line


def to_records(lines: Iterable[str]) -> Iterator[tuple[str, int]]:
    for line in lines:
        name, _, value = line.partition("=")
        yield name.strip(), int(value)


# ни одна строка ещё не прочитана — пайплайн только описан
pipeline = to_records(strip_and_skip(read_lines(Path("metrics.conf"))))
total = sum(v for _, v in pipeline)     # только здесь начинается работа

Ключевые свойства такого кода: память O(1) вместо O(N); первый результат доступен почти сразу (важно для стриминга и HTTP-ответов); каждая стадия тестируется отдельно на списке-заглушке; входом может быть что угодно итерируемое. Более общий взгляд на ленивость и потоки — в статье Ленивость и потоки трека функционального программирования.

itertools: словарь ленивых операций

from itertools import accumulate, chain, count, islice, pairwise, takewhile

# бесконечный поток + отсечение: первые 5 квадратов
print(list(islice((n * n for n in count(1)), 5)))     # [1, 4, 9, 16, 25]

# скользящие пары — разности между соседними замерами
temps = [10, 12, 11, 15]
print([b - a for a, b in pairwise(temps)])            # [2, -1, 4]

# нарастающий итог без ручного аккумулятора
print(list(accumulate([1, 2, 3, 4])))                 # [1, 3, 6, 10]

# ленивая склейка нескольких источников
print(list(chain([1, 2], (3, 4), {5})))               # [1, 2, 3, 4, 5]

# читаем, пока условие держится
print(list(takewhile(lambda x: x < 3, [1, 2, 3, 1]))) # [1, 2]

itertools.batched (3.12+) закрывает самую частую задачу продакшена — резать поток на пачки для батчевой вставки в БД или запроса к API:

from itertools import batched, islice

for chunk in batched(range(10), 3):
    print(chunk)
# (0, 1, 2)
# (3, 4, 5)
# (6, 7, 8)
# (9,)


def batched_compat(iterable, n):
    """Тот же приём для Python < 3.12."""
    it = iter(iterable)
    while chunk := tuple(islice(it, n)):
        yield chunk

Две мины в itertools, на которых регулярно подрываются:

• tee не бесплатен. Он буферизует элементы, которые прочитал один клон, но ещё не прочитал другой. Если один потребитель уйдёт вперёд на миллион элементов, буфер окажется в памяти целиком. Часто честнее прочитать вход дважды или материализовать список явно.
• groupby группирует только подряд идущие элементы — вход нужно сначала отсортировать тем же ключом. Кроме того, объект группы становится невалидным, как только вы перешли к следующей группе: list(g) нужно делать сразу.

from itertools import groupby
from operator import itemgetter

rows = [("b", 1), ("a", 2), ("b", 3)]
rows.sort(key=itemgetter(0))            # без сортировки будут три группы вместо двух
print({k: [v for _, v in g] for k, g in groupby(rows, key=itemgetter(0))})
# {'a': [2], 'b': [1, 3]}

Когда стандартного набора не хватает — есть more-itertools: chunked, windowed, unique_everseen, peekable, spy, first_true.

Контекстные менеджеры: детерминированное освобождение

with решает задачу, которую в C++ решает RAII, а в Java — try-with-resources: гарантировать, что парный «закрывающий» шаг выполнится при любом способе выхода из блока — обычном, через return, break, continue или исключение.

Обратите внимание на две детали, которые часто упускают. Первая: если __enter__ бросил исключение, __exit__ не будет вызван — ресурс ещё не считается захваченным, и убирать за собой должен сам __enter__. Вторая: возврат истинного значения из __exit__ гасит исключение. Написать return True «чтобы не падало» — верный способ годами терять ошибки; по умолчанию __exit__ должен возвращать None.

Класс против @contextmanager

import time
from contextlib import contextmanager


class Timer:
    """Менеджер классом: подходит, когда нужно состояние наружу."""

    def __init__(self, label: str) -> None:
        self.label = label
        self.elapsed = 0.0

    def __enter__(self) -> "Timer":
        self._t0 = time.perf_counter()
        return self                       # то, что попадёт в as

    def __exit__(self, exc_type, exc, tb) -> None:
        self.elapsed = time.perf_counter() - self._t0
        status = "ошибка" if exc_type else "ок"
        print(f"{self.label}: {self.elapsed:.3f} с ({status})")
        # возвращаем None — исключение, если было, летит дальше


@contextmanager
def timer(label: str):
    """То же самое генератором: код до yield — вход, после — выход."""
    t0 = time.perf_counter()
    try:
        yield                              # здесь выполняется тело with
    finally:                               # ОБЯЗАТЕЛЬНО: иначе при исключении не сработает
        print(f"{label}: {time.perf_counter() - t0:.3f} с")

Правило выбора простое: @contextmanager — для линейной логики «подготовил → отдал → прибрал»; класс — когда менеджер должен быть переиспользуемым, реентерабельным или хранить состояние, доступное после выхода.

Три грабли @contextmanager

1. Забытый try/finally. Без него код после yield не выполнится при исключении в теле — а это ровно тот случай, ради которого with и придуман.

@contextmanager
def broken(conn):
    tx = conn.begin()
    yield tx
    tx.commit()        # при исключении в теле НЕ выполнится, транзакция повиснет

2. Одноразовость. Объект, возвращённый @contextmanager-функцией, содержит один конкретный генератор. Повторный with над той же переменной падает:

cm = timer("шаг")
with cm: pass
with cm: pass      # RuntimeError: generator didn't yield

Правильно — вызывать фабрику заново: with timer("шаг"):. Класс Timer этой проблемы не имеет. Это же различие описано в документации как «reusable» и «reentrant» менеджеры: suppress() и redirect_stdout() реентерабельны, обёртки @contextmanager — нет.

3. Проглоченное исключение. Если внутри @contextmanager вы ловите исключение и не пробрасываете его, оно исчезнет — ровно как return True из __exit__.

@contextmanager
def transaction(conn):
    conn.execute("BEGIN")
    try:
        yield conn
    except BaseException:      # BaseException, а не Exception: откат нужен и при KeyboardInterrupt
        conn.rollback()
        raise                  # без raise транзакция «успешно» съест ошибку
    else:
        conn.commit()

ExitStack: когда число ресурсов известно только в рантайме

from contextlib import ExitStack
from itertools import chain
from pathlib import Path


def merge(paths: list[Path], out: Path) -> None:
    """Открыть N файлов сразу — и гарантированно закрыть все, даже если пятый не открылся."""
    with ExitStack() as stack:
        files = [stack.enter_context(p.open(encoding="utf-8")) for p in paths]
        target = stack.enter_context(out.open("w", encoding="utf-8"))
        stack.callback(print, "слияние завершено")   # произвольный колбэк на выход
        for line in chain.from_iterable(files):      # ленивая склейка всех потоков
            target.write(line)

Ещё два приёма из того же модуля: stack.pop_all() — «передать владение» ресурсами наружу, отменив автоматическую очистку (нужно, когда функция возвращает открытый ресурс); AsyncExitStack — то же для async with.

Малый инвентарь contextlib

import sys
from contextlib import nullcontext, suppress
from pathlib import Path

# suppress вместо try/except/pass — сразу видно, ЧТО именно гасим
with suppress(FileNotFoundError):
    Path("cache.tmp").unlink()

# ресурс нужен не всегда, но тело блока остаётся одним и тем же
path: Path | None = None
ctx = path.open(encoding="utf-8") if path else nullcontext(sys.stdin)
with ctx as source:
    for line in source:
        print(line, end="")

Функции, декорированные @contextmanager, наследуют ContextDecorator, поэтому их можно применять и как декоратор — и в этой роли они уже переиспользуемы, потому что на каждый вызов создаётся новый генератор:

@timer("обработка")
def handle(batch):     # эквивалент with timer("обработка") внутри функции
    ...

Асинхронные близнецы

from contextlib import asynccontextmanager, aclosing


@asynccontextmanager
async def session(url: str):
    conn = await connect(url)
    try:
        yield conn
    finally:
        await conn.close()


async def stream(url: str):
    async with session(url) as conn:            # __aenter__ / __aexit__
        async with aclosing(conn.rows()) as rows:
            async for row in rows:              # __aiter__ / __anext__
                yield row                       # это асинхронный генератор

Асинхронные comprehension-ы (PEP 530) тоже существуют: [x async for x in agen if await ok(x)]. Подробности — в конкурентности.

Практика: ETL-пайплайн целиком

Соберём всё вместе: ленивое чтение нескольких gzip-CSV, парсинг с пропуском битых строк, фильтр, батчинг и транзакционная вставка. Память константна независимо от размера входа.

from __future__ import annotations

import csv
import gzip
import logging
import sqlite3
from collections.abc import Iterable, Iterator
from contextlib import ExitStack, closing, contextmanager
from dataclasses import dataclass
from itertools import batched
from pathlib import Path

log = logging.getLogger(__name__)


@dataclass(frozen=True, slots=True)
class Event:
    user_id: int
    kind: str
    amount: int


def read_rows(paths: Iterable[Path]) -> Iterator[dict[str, str]]:
    """Склеивает несколько .csv.gz в один ленивый поток словарей."""
    for path in paths:
        # with внутри генератора: файл закроется и при обычном исходе,
        # и при .close() генератора (GeneratorExit долетит до finally)
        with gzip.open(path, "rt", newline="", encoding="utf-8") as fh:
            yield from csv.DictReader(fh)


def parse(rows: Iterable[dict[str, str]]) -> Iterator[Event]:
    """Битая строка не роняет пайплайн — логируется и пропускается."""
    for lineno, row in enumerate(rows, start=1):
        try:
            yield Event(int(row["user_id"]), row["kind"], int(row["amount"]))
        except (KeyError, ValueError) as exc:
            log.warning("строка %d пропущена: %s", lineno, exc)


def only_paid(events: Iterable[Event]) -> Iterator[Event]:
    return (e for e in events if e.kind == "paid" and e.amount > 0)


@contextmanager
def transaction(conn: sqlite3.Connection) -> Iterator[sqlite3.Connection]:
    conn.execute("BEGIN")
    try:
        yield conn
    except BaseException:
        conn.rollback()
        raise
    else:
        conn.commit()


def load(db: Path, paths: Iterable[Path], batch_size: int = 1000) -> int:
    total = 0
    with ExitStack() as stack:
        conn = stack.enter_context(
            closing(sqlite3.connect(db, isolation_level=None))
        )
        conn.execute("CREATE TABLE IF NOT EXISTS paid(user_id INT, amount INT)")

        pipeline = only_paid(parse(read_rows(paths)))     # три ленивые стадии
        for chunk in batched(pipeline, batch_size):        # пачка кортежей
            with transaction(conn):
                conn.executemany(
                    "INSERT INTO paid VALUES (?, ?)",
                    [(e.user_id, e.amount) for e in chunk],
                )
            total += len(chunk)
    return total

Что здесь идиоматично и почему это важно: каждая стадия принимает Iterable и возвращает Iterator, поэтому в тестах вместо файлов подставляется список словарей; ресурсы управляются with, а не try/finally вручную; батчинг отделён от бизнес-логики; ошибки парсинга обрабатываются на своём уровне, а не глушатся общим except.

Малые идиомы, из которых складывается стиль

Редкая, но полезная конструкция — for ... else: блок else выполняется, если цикл завершился без break. Читается плохо, но заменяет флаг found:

for user in users:
    if user.is_admin:
        break
else:
    raise RuntimeError("в списке нет ни одного администратора")

Честно: где ленивость и протоколы проигрывают

Идиоматичность не бесплатна, и продавать её как серебряную пулю нечестно.

• Числовые данные. Генератор обрабатывает элемент за ~50–150 нс интерпретаторных накладных. Для массива на 10⁸ чисел это десятки секунд там, где numpy справится за доли секунды векторизованно. Правило: как только данные однородные и числовые — переходите на numpy/pandas (см. работу с данными), а генераторы оставляйте на границе ввода-вывода.
• Отладка. Traceback показывает точку потребления, а не место, где пайплайн был собран. Стек из пяти вложенных генераторов читается тяжело, а точка останова в отладчике срабатывает в неожиданном порядке.
• Отложенные исключения. Ошибка в первой стадии проявится где-то в sum() глубоко внизу, возможно, после того как транзакция уже открыта. Ленивость перемешивает порядок эффектов — с ресурсами это опасно.
• Нет len, нет индексации, нет повторного обхода. Любой код вида if len(data) > 0 ломается, стоит подменить список генератором. Аннотация Iterable[T] — это честное обещание «только один проход».
• Нет параллелизма. Генератор — однопоточный автомат. Ускорить пайплайн можно только вынеся стадии в процессы через очереди; сам по себе он не масштабируется.
• Скрытый расход. tee, groupby без сортировки, sorted в середине «ленивой» цепочки — всё это молча материализует данные и обнуляет выигрыш.

Выигрывает же Python здесь в том, чего нет у большинства языков: один протокол описывает всё — файл, сокет, курсор БД, HTTP-стрим, Kafka-консьюмер, генератор, диапазон. Функция, написанная под Iterable[str], работает со всеми ними без единой правки. Это и есть главная причина, по которой Python так хорош на «склеивающем» слое систем.

Типичные ошибки — сводная таблица

Чек-лист код-ревью

• Функция принимает Iterable, а не list, если ей нужен только обход?
• Ленивое значение не обходится дважды и не попадает в len?
• Каждый @contextmanager содержит try/finally (или except/else) вокруг yield?
• Ни один __exit__ не возвращает истину без явного намерения проглотить исключение?
• Comprehension помещается в одну-две строки и не имеет побочных эффектов?
• Ресурсы внутри генератора закрываются через with, а сам генератор — через closing, если его могут не дочитать?
• Пачки для БД/API режутся batched, а не ручным аккумулятором со сбросом?
• Там, где данные числовые и однородные, вместо генераторов используется numpy?
• В цепочке нет скрытой материализации (sorted, tee, list) в неожиданном месте?

Мини-итог

Идиоматичный Python — это код, написанный на языке протоколов. for — это iter + next + StopIteration, и из этого следует всё остальное: одноразовость итераторов, отсутствие len, возможность бесконечных потоков. Генератор — функция с замороженным кадром, чьё состояние есть точка выполнения; send/throw/close делают из него полноценную сопрограмму, а yield from — композицию. Comprehension — выражение, описывающее результат; скобки решают, материализуется он или течёт. with — единственный надёжный способ гарантировать освобождение ресурса, а contextlib превращает в контекстный менеджер почти что угодно. Плата за ленивость — сложность отладки, отложенные эффекты и интерпретаторные накладные на элемент; там, где данные числовые, её платить не нужно.

Источники

• Iterator Types и Generator Types — формальное описание протоколов и методов генератора.
• PEP 255 — Simple Generators, PEP 342 — Coroutines via Enhanced Generators, PEP 380 — Syntax for Delegating to a Subgenerator, PEP 479 — Change StopIteration handling.
• PEP 289 — Generator Expressions, PEP 572 — Assignment Expressions, PEP 709 — Inlined comprehensions.
• PEP 343 — The “with” Statement, PEP 525 — Asynchronous Generators, PEP 530 — Asynchronous Comprehensions.
• contextlib — ExitStack, suppress, closing, aclosing, различие reusable/reentrant.
• itertools — включая раздел «Itertools Recipes» с готовыми функциями.
• Functional Programming HOWTO — официальный обзор итераторов и генераторов.
• Luciano Ramalho, Fluent Python, 2nd ed. — главы 17 (итераторы и генераторы) и 18 (with, match).
• David Beazley, Generator Tricks for Systems Programmers — каноническое введение в пайплайны на генераторах.
• Brett Slatkin, Effective Python, 2nd ed. — пункты о comprehension-ах, генераторах и contextlib.
• more-itertools — расширенный набор ленивых рецептов.

Что дальше

Аннотации типов: mypy, Protocol, generics, строгость в реальном проекте — как описать протоколы, которые мы только что разобрали, в системе типов: чем Iterable[T] отличается от Iterator[T] и Generator[Y, S, R], зачем нужны Protocol и вариантность, и как включить строгий mypy в проекте, который писался без типов.