Генераторы — способ писать итераторы просто и эффективно. Они отдают элементы по требованию, сберегают память и делают код короче. В основе — оператор yield, который приостанавливает выполнение функции и сохраняет её локальное состояние до следующего вызова next(). В этом расширенном руководстве вы разберёте базовый синтаксис и протокол итератора, генераторные выражения, конвейеры (pipeline), управление генератором через send/throw/close, делегирование yield from, тестирование и производительность, типичные ошибки, работу с файлами и сетевыми потоками, а также интеграцию с itertools и приёмы для production-кода.
1. Что такое генератор и зачем он нужен
Генератор — функция, внутри которой есть yield. Каждый вызов next() продолжает исполнение до ближайшего yield, возвращая очередной элемент и «замораживая» стек. Это позволяет строить ленивые конвейеры обработки данных и обходиться без промежуточных списков.
def countdown(n):
while n > 0:
yield n
n -= 1
it = countdown(3)
print(next(it)) # 3
print(next(it)) # 2
print(list(it)) # [1] - Ленивость: вычисление происходит в момент потребления.
- Экономия памяти: держим в памяти только текущий элемент.
- Простота: протокол итератора реализован «сам собой» без классов.
2. Разница между return и yield
return завершает функцию один раз и навсегда, возвращая единственное значение. yield «режет» результат на порции — сколько запросили, столько и отдали. Если в генераторе сделать return value, это приведёт к завершению с StopIteration(value) — полезно при делегировании через yield from.
def once():
yield 1
return 99 # значение окажется в StopIteration.value
g = once()
print(next(g)) # 1
try:
next(g)
except StopIteration as e:
print(e.value) # 99 3. next(), завершение и StopIteration
Цикл for сам вызывает next() и ловит StopIteration. Вручную это выглядит так:
def pair():
yield "A"
yield "B"
g = pair()
print(next(g)) # A
print(next(g)) # B
try:
print(next(g)) # генератор исчерпан
except StopIteration:
print("done") Правило PEP 479: не поднимайте StopIteration внутри генератора вручную для выхода — используйте return. Явный StopIteration в теле генератора может быть преобразован в RuntimeError.
4. Генератор вместо класса-итератора
Тот же эффект можно получить классом с __iter__ и __next__, но генератор проще:
# Классический итератор
class Squares:
def __init__(self, n):
self.n = n
self.i = 1
def __iter__(self):
return self
def __next__(self):
if self.i > self.n:
raise StopIteration
v = self.i * self.i
self.i += 1
return v
print(list(Squares(5)))
# Эквивалент генератором
def squares(n):
i = 1
while i <= n:
yield i * i
i += 1
print(list(squares(5))) 5. Генераторные выражения и списковые включения
Списковое включение создаёт список немедленно. Генераторное выражение создаёт итератор, который выдаёт элементы по мере надобности.
# Список: память ∝ числу элементов
data = [x*x for x in range(1_000_000)]
# Генератор: память почти константная
data_gen = (x*x for x in range(1_000_000))
print(sum(data_gen)) # потоковое потребление - Если результат нужен целиком и многократно — список уместнее.
- Если нужен одноразовый проход — генератор экономит ресурсы.
6. Ленивые вычисления на больших данных
Генераторы особенно полезны для больших файлов и потоков. Вы читаете по строке, фильтруете, преобразуете и агрегируете без промежуточных структур.
def read_lines(path):
with open(path, "r", encoding="utf-8") as f:
for line in f:
yield line.rstrip("n")
def grep(lines, needle):
for line in lines:
if needle in line:
yield line
def to_fields(lines, sep=";"):
for line in lines:
yield line.split(sep)
for fields in to_fields(grep(read_lines("app.log"), "ERROR")):
print(fields[:3]) 7. Конвейеры (pipeline) из генераторов
Стройте небольшие одноцелевые генераторы и комбинируйте их:
def numbers():
i = 0
while True:
yield i
i += 1
def only_even(it):
for x in it:
if x % 2 == 0:
yield x
def square(it):
for x in it:
yield x * x
pipe = square(only_even(numbers()))
for x in pipe:
if x > 100:
break
print(x) Достоинство такого подхода — читаемость и лёгкая тестируемость каждого звена.
8. Управление генератором: send, throw, close
Генератор поддерживает двустороннее взаимодействие. Через send(value) можно передать значение в точку последнего yield. throw(exc) вбрасывает исключение внутрь. close() завершает генератор с GeneratorExit.
def accumulator():
total = 0
try:
while True:
x = (yield total) # получаем число через send
if x is None:
break
total += x
finally:
# здесь можно освободить ресурсы
pass
g = accumulator()
print(next(g)) # 0
print(g.send(10)) # 10
print(g.send(5)) # 15
g.close() Не подавляйте GeneratorExit и не делайте в блоке finally долгих операций — генератор закрывается как часть механизма очистки.
9. Делегирование: yield from и возврат значения
yield from проксирует итерацию и методы управления в подгенератор, а также возвращает его итоговое значение (через return в подгенераторе).
def sub():
total = 0
while True:
x = (yield)
if x is None:
return total
total += x
def super_acc():
# получим финальный итог из подгенератора
result = yield from sub()
yield result
g = super_acc()
next(g)
g.send(3)
g.send(7)
print(g.send(None)) # 10 10. Интеграция с itertools
Модуль itertools идеально сочетается с генераторами: count, cycle, repeat, islice, takewhile, dropwhile, chain, compress, groupby, accumulate, tee, zip_longest, product, permutations, combinations.
from itertools import islice, takewhile, accumulate, chain
def naturals():
i = 1
while True:
yield i
i += 1
squares = (x*x for x in naturals())
first_ten = list(islice(squares, 10))
print(first_ten)
prefix_sums = list(islice(accumulate(naturals()), 5))
print(prefix_sums)
print(list(chain([0], [1, 2], range(3, 6)))) tee делает «копии» итератора, но имейте в виду, что одна копия может буферизовать элементы для другой — это может вылиться в заметное потребление памяти.
11. Рецепты генераторов для production
11.1. Скользящее окно
from collections import deque
def sliding_window(it, size):
buf = deque(maxlen=size)
for x in it:
buf.append(x)
if len(buf) == size:
yield tuple(buf)
print(list(sliding_window(range(1, 7), 3))) 11.2. Пакетирование (batch) потока
def batched(it, n):
batch = []
for x in it:
batch.append(x)
if len(batch) == n:
yield batch
batch = []
if batch:
yield batch
for b in batched(range(1, 8), 3):
print(b) 11.3. Уникализация с сохранением порядка
def dedupe(it, key=None):
seen = set()
for x in it:
k = x if key is None else key(x)
if k not in seen:
seen.add(k)
yield x
print(list(dedupe(["a","b","a","c","b"]))) 11.4. Слияние отсортированных потоков
import heapq
def merge_sorted(*iters):
heap = []
for it_index, it in enumerate(map(iter, iters)):
try:
first = next(it)
heap.append((first, it_index, it))
except StopIteration:
pass
heapq.heapify(heap)
while heap:
val, idx, it = heapq.heappop(heap)
yield val
try:
nxt = next(it)
heapq.heappush(heap, (nxt, idx, it))
except StopIteration:
pass
print(list(merge_sorted([1,4,9], [2,6], [0,3,7,8]))) 11.5. Потоковый разбор CSV
import csv
def read_csv(path, encoding="utf-8"):
with open(path, newline="", encoding=encoding) as f:
rdr = csv.DictReader(f)
for row in rdr:
yield row
def select(it, *cols):
for row in it:
yield {c: row[c] for c in cols}
def where(it, pred):
for row in it:
if pred(row):
yield row
rows = where(select(read_csv("users.csv"), "id", "age"), lambda r: int(r["age"]) >= 18)
for r in rows:
print(r) 11.6. Хвост файла (tail -f)
import time
def follow(path):
with open(path, "r", encoding="utf-8") as f:
f.seek(0, 2) # в конец
while True:
line = f.readline()
if not line:
time.sleep(0.2)
continue
yield line.rstrip("n")
for line in follow("app.log"):
if "ERROR" in line:
print(line) 11.7. Ограничитель скорости (rate limiter)
import time
def rate_limit(it, per_sec):
interval = 1.0 / per_sec
last = 0.0
for x in it:
now = time.time()
delta = now - last
if delta < interval:
time.sleep(interval - delta)
yield x
last = time.time()
for x in rate_limit(range(5), per_sec=2):
print(x, time.time()) 11.8. Экспоненциальный бэкофф для повторов
import random, time
def retries(max_tries=5, base=0.1, factor=2.0):
delay = base
for attempt in range(1, max_tries + 1):
yield attempt, delay
delay *= factor
def unreliable():
return random.random() < 0.3 # 30% «успех»
for attempt, delay in retries():
if unreliable():
print("ok on try", attempt)
break
print("fail, sleep", delay)
time.sleep(delay) 12. Контекстные менеджеры на основе генераторов
Декоратор contextlib.contextmanager позволяет писать контекстные менеджеры как генераторы с одним yield: всё до yield — «вход», всё после — «выход».
from contextlib import contextmanager
@contextmanager
def opened(path, mode="r", encoding="utf-8"):
f = open(path, mode, encoding=encoding)
try:
yield f
finally:
f.close()
with opened("file.txt") as f:
print(f.readline()) 13. Аннотации типов для генераторов
Используйте typing.Generator[Y, S, R], где Y — тип выдаваемых значений, S — тип, передаваемый через send, R — тип «return» (доступен как StopIteration.value при делегировании).
from typing import Generator
def gen() -> Generator[int, None, None]:
yield 1
yield 2 14. Тестирование и измерение
Генераторы легко тестировать через материализацию небольших кусочков, а производительность — через timeit.
import timeit
setup = "nums = range(200_000)"
code_list = "sum([x*x for x in nums])"
code_gen = "sum(x*x for x in nums)"
print("list:", timeit.timeit(code_list, setup=setup, number=5))
print("gen :", timeit.timeit(code_gen, setup=setup, number=5)) Идея: если вы в итоге всё равно строите список целиком и дважды проходите по нему — генератор преимущества не даст. Но в одноразовых потоках он выигрывает.
15. Частые ошибки и подводные камни
- Неожиданная материализация. Вызов list(gen) или sorted(gen) загружает весь поток в память.
- Повторное использование. Исчерпанный генератор «пуст». Создавайте новый экземпляр.
- StopIteration внутри тела. Для выхода используйте return, а не raise StopIteration.
- Выход из finally слишком тяжёлый. Не задерживайте закрытие генератора долгими операциями.
- tee без оглядки на память. Понимайте, что одна «ветка» может буферизовать данные для другой.
- Смешение ответственности. Генератор должен делать одну небольшую вещь: читать, фильтровать, преобразовывать или агрегировать.
16. Сравнение с асинхронными генераторами
У обычных генераторов next() блокирующий. Асинхронные генераторы (async def с yield) работают с async for и выдают элементы по мере готовности асинхронных операций. Если у вас I/O-потоки и нужен неблокирующий код — изучайте async for и async generators. Принципы ленивости и композиции схожи, но протоколы другие.
17. Расширённые приёмы диагностики
У генератора есть атрибуты introspection: gi_frame (текущий фрейм), gi_running (исполняется ли сейчас), gi_code (объект кода). Они полезны при отладке сложных конвейеров, но в проде лучше не полагаться на них.
def gen():
yield 1
yield 2
g = gen()
print(g.gi_code.co_name) # 'gen'
print(g.gi_running) # False
print(next(g)) # 1 18. Расширенные примеры, проверенные временем
18.1. Группировка по ключу (похожие на groupby)
def group_by_sorted(it, key):
it = iter(it)
try:
first = next(it)
except StopIteration:
return
cur_key = key(first)
bucket = [first]
for x in it:
k = key(x)
if k != cur_key:
yield cur_key, bucket
cur_key = k
bucket = [x]
else:
bucket.append(x)
yield cur_key, bucket
data = ["a1","a2","b9","b3","c0"]
print(list(group_by_sorted(data, key=lambda s: s[0]))) 18.2. Разворачивание (flatten) вложенных списков
def flatten(nested):
for x in nested:
if isinstance(x, (list, tuple)):
yield from flatten(x)
else:
yield x
print(list(flatten([1,[2,[3,4]],5]))) 18.3. Генератор мини-парсера строк
def tokens(s, seps=" ,;"):
buf = []
for ch in s:
if ch in seps:
if buf:
yield "".join(buf)
buf.clear()
else:
buf.append(ch)
if buf:
yield "".join(buf)
print(list(tokens("a, b; c d"))) 19. Мини-шпаргалка
- Базовый генератор: функция с yield.
- Протокол: next() выдаёт значение, исчерпание — StopIteration.
- Генераторное выражение: (expr for x in it if cond).
- Композиция: объединяйте простые генераторы в конвейер.
- Управление: send, throw, close, делегирование yield from.
- Чистый дизайн: маленькие функции, минимум побочных эффектов.
- Производительность: не материализуйте поток без нужды; измеряйте timeit.
Генераторы позволяют писать быстрый и прозрачный код для потоков данных: логов, файлов, сетевых ответов, бесконечных последовательностей. Освоив yield, генераторные выражения, yield from и приёмы композиции, вы получите инструмент, который масштабируется от коротких скриптов до production-пайплайнов и ETL-процессов.