Генераторы в Python

Введение: что такое генератор и зачем он нужен

Генератор инкапсулирует протокол итератора (итерабельность и **next**) и выдаёт элементы по одному. Он естественно поддерживает ленивые вычисления: пока вы не попросили следующий элемент, он не вычисляется. Выгоды: экономия памяти, возможность работать с бесконечными или очень длинными последовательностями, потоками данных и большими файлами, простая композиция этапов обработки через pipeline, более чистый код по сравнению с ручной реализацией класса-итератора.

Ключевые схемы использования: чтение файлов построчно, фильтрация и агрегация без промежуточных структур, потоковая нормализация и дедупликация, веб-скрапинг и обработка очередей, «ленивые» преобразования (map/filter) и генерация последовательностей под запрос.

Генераторы vs функции и итераторы

• Функция + return возвращает результат один раз и завершает выполнение.
• Генератор + yield отдаёт значения многократно, «ставя на паузу» выполнение и сохраняя локальное состояние между вызовами next().
• Итератор-класс требует писать **iter**/**next** и вручную бросать StopIteration; генератор делает то же самое декларативно.
• for в Python сам вызывает next() и ловит StopIteration — поэтому генераторы особенно удобны в циклах.

# return vs yield

def build_list(n):
out = []
for i in range(n):
out.append(i*i)
return out # материализация всей коллекции

def gen_squares(n):
for i in range(n):
yield i*i # отдаём по одному, без списка

Синтаксис yield: пауза выполнения и сохранение состояния

yield — инструкция, которая отдаёт значение наружу и замораживает выполнение до следующего запроса. Следующий next() продолжает исполнять функцию сразу после yield. В генераторе может быть много yield. Важно: использовать return внутри генератора можно, но без значения (или с значением для передачи наружу через StopIteration.value в обёртках, см. ниже). Начиная с Python 3.7, прямое возбуждение StopIteration из тела генератора приводит к RuntimeError — корректно завершать генератор нужно return.

def countdown(n: int):
while n > 0:
yield n # отдаём текущее значение и «замеряем»
n -= 1

def demo_return_value():
# вернём итог через return, его можно поймать на верхнем уровне (см. yield from)
total = 0
for i in range(3):
total += i
yield i
return total # корректный способ завершения генератора с «результатом»

Итерация: next(), завершение и StopIteration

next() продвигает генератор к следующему yield. Когда значения кончились, next() возбуждает StopIteration, что для цикла for — нормальный сигнал завершения. Это поведение одинаково и для обычных итераторов, и для генераторов.

gen = countdown(3)
print(next(gen)) # 3
print(next(gen)) # 2
print(next(gen)) # 1
try:
print(next(gen)) # StopIteration
except StopIteration as e:
print("Готово")

Генератор как упрощённый итератор: класс vs функция

Эквивалентная логика на классах многословна, требует хранить состояние и своевременно бросать StopIteration. Генераторная функция выражает то же самое кратко и безопасно, облегчая сопровождение.

# Класс-итератор (шаблонный «длинный» вариант)
class RangeIterator:
def __init__(self, start, stop, step=1):
self.cur, self.stop, self.step = start, stop, step
def __iter__(self):
return self
def __next__(self):
if self.cur >= self.stop:
raise StopIteration
val = self.cur
self.cur += self.step
return val

# Генераторная функция — эквивалент короче и выразительнее

def range_gen(start, stop, step=1):
cur = start
while cur < stop:
yield cur
cur += step

Базовые примеры генераторов

Набор типичных шаблонов: последовательности, обход коллекций, «уплощение» вложенных структур, бесконечные потоки.

def squares(n: int):
for x in range(n):
yield x * x

def walk_dict(d: dict):
for k, v in d.items():
yield k, v

def flatten(nested):
for seq in nested:
for item in seq:
yield item

def naturals(start=0):
while True:
yield start
start += 1

Генераторные выражения vs list comprehension

Генераторное выражение возвращает поток (итератор), а списковое включение сразу строит список в памяти. Если нужен весь результат и случайный доступ — берите список. Если важны поток, экономия памяти и цепочки обработки — генераторное выражение. Оба варианта изолируют переменные цикла внутри своей области видимости.

# список — материализация
evens_list = [x for x in range(10_000) if x % 2 == 0]
# генератор — поток
evens_iter = (x for x in range(10_000) if x % 2 == 0)

# практический пример: считаем сумму чётных — поток не создаёт большой список

total = sum(x for x in range(1_000_000) if x % 2 == 0)

Ленивые вычисления и большие файлы

Чтение и обработка «на лету» снимают пиковые нагрузки по памяти. Для «тяжёлых» логов, CSV/JSONL и потоков данных это критично. Добавляйте якоря производительности: избегайте промежуточных списков, используйте вычисления на проходе, агрегируйте на лету.

from pathlib import Path

def read_lines(path: str):
with open(path, "r", encoding="utf-8") as f:
for line in f:
yield line.rstrip("\n")

def grep(substr: str, lines):
for line in lines:
if substr in line:
yield line

def to_ints(lines):
for line in lines:
try:
yield int(line)
except ValueError:
continue

# pipeline: чтение -> фильтрация -> преобразование -> первые N

def take(n: int, it):
for i, x in enumerate(it, 1):
if i > n:
return
yield x

for row in take(5, to_ints(grep("ERROR", read_lines("app.log")))):
print(row)

Конвейеры генераторов (pipeline)

Связывайте простые генераторы в цепочки: фильтрация, преобразование, агрегация — каждое звено делает одну вещь. Такой pipeline повышает читаемость и переиспользуемость, снижает связанность, упрощает тестирование.

def find_prime(limit=100):
num = 2
while num <= limit:
for p in range(2, int(num ** 0.5) + 1):
if num % p == 0:
break
else:
yield num
num += 1

def only_odd(seq):
for n in seq:
if n % 2 != 0:
yield n

def square(seq):
for n in seq:
yield n*n

for n in take(10, square(only_odd(find_prime(1000)))):
print(n)

Практические кейсы: файлы, фильтрация, веб-скрапинг

Шаблоны: «чтение файлов построчно», потоковая фильтрация, разбиение на батчи для записи или сетевых вызовов, веб-скрапинг (постраничные итерации), обработка очередей.

def batched(it, size: int):
batch = []
for item in it:
batch.append(item)
if len(batch) == size:
yield batch
batch = []
if batch:
yield batch

def parse_csv_lines(lines):
for line in lines:
cols = [c.strip() for c in line.split(",")]
yield cols

# Пример: чтение -> парсинг -> батчи -> обработка

for pack in batched(parse_csv_lines(read_lines("data.csv")), size=1000):
process(pack)

Расширенные приёмы: send, throw, close

Помимо next(), у генератора есть API управления. send(value) отправляет значение в точку последнего yield. throw(exc) возбуждает исключение внутри генератора. close() корректно завершает генератор, вызывая GeneratorExit. Полезно для «сопрограмм-подобного» взаимодействия и тонкого контроля.

def averager():
total = 0.0
count = 0
avg = None
try:
while True:
x = yield avg # сюда «прилетает» send(x)
total += x
count += 1
avg = total / count
except GeneratorExit:
# можно освободить ресурсы
return

g = averager()
next(g) # праймим генератор
print(g.send(10)) # 10.0
print(g.send(20)) # 15.0
g.close()

yield from: делегирование и возврат значения подгенератора

yield from делегирует итерацию подгенератору, убирая вложенные циклы; кроме того, он позволяет получить «возвратное» значение подгенератора (StopIteration.value) как результат выражения yield from. Это упрощает агрегирующие сценарии.

def subgen():
total = 0
for x in range(5):
yield x
total += x
return total # вернём агрегат

def wrapper():
result = yield from subgen() # здесь окажется return из subgen
yield ("sum", result)

print(list(wrapper())) # [0, 1, 2, 3, 4, ('sum', 10)]

Ошибки и отладка

• StopIteration — нормальный финал при лишнем next(); в тестах оборачивайте вызов в try/except.
• PEP 479: прямой raise StopIteration внутри генератора превращается в RuntimeError. Используйте return для завершения.
• TypeError: can't send non-None value to a just-started generator — сначала вызовите next(g) перед первым send().
• ValueError: generator already executing — не реентерабельный вызов; не запускайте один генератор параллельно.
• Случайная материализация: не оборачивайте поток в list() без необходимости.

def safe_next(it):
try:
return next(it)
except StopIteration:
return None

Тестирование и производительность: timeit, память, асимптотика

Измеряйте время с timeit, а память — с tracemalloc или грубо по размеру промежуточных структур. Важнее всего асимптотика памяти: генераторные выражения и pipeline не создают большие коллекции и стабильно работают на длинных потоках.

import timeit, tracemalloc

setup = "data = list(range(200_000))"
gen_stmt = "sum(x for x in data if x % 2 == 0)"
list_stmt = "sum([x for x in data if x % 2 == 0])"

print("gen:", timeit.timeit(gen_stmt, setup=setup, number=10))
print("list:", timeit.timeit(list_stmt, setup=setup, number=10))

tracemalloc.start()
sum(x for x in range(1_000_000))
cur, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
print("peak bytes:", peak)

Лучшие практики: читаемость, композиция, аннотации

• Одна генераторная функция — одна задача; собирайте конвейер из мелких блоков.
• Именуйте по действию: read_lines, filter_errors, chunks, only_odd.
• Не копите в память; применяйте генераторные выражения внутри sum/any/all/max/min.
• Покрывайте генераторы тестами на граничные случаи: пустой поток, один элемент, большие объёмы.
• Используйте аннотации: from typing import Iterator, Iterable, Generator. Для сложных случаев тип Generator[YieldType, SendType, ReturnType].

from typing import Generator, Iterable

def moving_avg(xs: Iterable[float], k: int) -> Generator[float, None, None]:
buf = []
s = 0.0
for x in xs:
buf.append(x); s += x
if len(buf) > k:
s -= buf.pop(0)
if len(buf) == k:
yield s / k

Итоги и мини-шпаргалка

Коротко: yield отдаёт значение и ставит выполнение на паузу; next() продолжает; завершение — StopIteration. Генератор — компактная альтернатива класс-итератору, оптимален для ленивых вычислений и больших данных. Генераторные выражения дают поток вместо списка. Конвейеры из генераторов — простой путь к масштабируемой обработке: «читать → фильтровать → преобразовывать → агрегировать». Для расширенных сценариев используйте send/throw/close и yield from для делегирования и получения возвращаемого значения подгенератора. Тестируйте с timeit, мониторьте память, избегайте лишней материализации и держите функции маленькими и переиспользуемыми — так код останется быстрым и предсказуемым.