Итераторы и генераторы

Итераторы и генераторы в Python

Маршрут: Инженерная основа · тема 1 из 8

Навыки: M3

До урока: control_flow, functions, python_testing.

Результат: ленивые потоковые конвейеры с контролируемым потреблением памяти.

Основа: итерируемые объекты, итераторы, yield и генераторные выражения. Работа в эксплуатации: потоковая обработка, ограниченная память и корректное закрытие. Углубление: send, throw, close и состояние кадра генератора.

Подтверждение: конвейер большого входа и тест верхней границы памяти.

Итератор выдаёт элементы по запросу. Генератор хранит состояние вычисления, но не обязан хранить все результаты; это позволяет строить потоковые и даже бесконечные последовательности при условии, что потребитель ограничивает обход.

#1. Протокол итерации

Два ключевых понятия:

Iterable (итерируемый): объект с методом __iter__(), возвращающим итератор. Примеры: list, str, dict, set, range.
Iterator (итератор): объект с методами __iter__() и __next__(). __next__() возвращает следующий элемент или вызывает StopIteration.

lst = [1, 2, 3]       # iterable
it = iter(lst)        # iterator = lst.__iter__()
print(next(it))       # 1  — вызывает it.__next__()
print(next(it))       # 2
print(next(it))       # 3
next(it)              # StopIteration!

Ключевое различие:

Итерируемый можно обойти много раз: for x in lst: ... можно вызвать снова.
Итератор — одноразовый: исчерпан — больше не работает.

it = iter([1, 2])
list(it)  # [1, 2]
list(it)  # [] — итератор уже исчерпан!
iter(it) is it  # True — iter() не сбрасывает исчерпанный итератор

#Собственный итератор

class Countdown:
    def __init__(self, n):
        self.n = n

    def __iter__(self):
        return self          # итератор возвращает self

    def __next__(self):
        if self.n <= 0:
            raise StopIteration
        val = self.n
        self.n -= 1
        return val

for i in Countdown(3):
    print(i)  # 3, 2, 1

# Можно создать отдельный класс-итератор и iterable:
class NumberRange:
    def __init__(self, start, end):
        self.start = start
        self.end = end

    def __iter__(self):
        return NumberRangeIterator(self.start, self.end)

class NumberRangeIterator:
    def __init__(self, current, end):
        self.current = current
        self.end = end

    def __iter__(self):
        return self

    def __next__(self):
        if self.current >= self.end:
            raise StopIteration
        val = self.current
        self.current += 1
        return val

r = NumberRange(1, 4)
print(list(r))  # [1, 2, 3]
print(list(r))  # [1, 2, 3] — снова, потому что r — iterable, не iterator

#`iter()` в двухаргументной форме

import io
buf = io.StringIO("line1\nline2\nSTOP\nline3\n")
# iter(callable, sentinel): вызывает callable до тех пор, пока не вернёт sentinel
for line in iter(buf.readline, "STOP\n"):
    print(line.strip())  # line1, line2 — останавливается на STOP

#2. Генераторы через `yield`

Функция с yield превращается в генераторную функцию — вызов создаёт объект-генератор.

def fibonacci(n):
    a, b = 0, 1
    for _ in range(n):
        yield a
        a, b = b, a + b

gen = fibonacci(5)     # объект-генератор создан, код ещё не выполнялся
print(type(gen))       # <class 'generator'>
print(next(gen))       # 0 — выполняется до первого yield
print(next(gen))       # 1
print(list(gen))       # [1, 2, 3] — остаток

Что происходит при yield:

Возвращает значение вызывающему
Сохраняет состояние фрейма генератора: локальные переменные и позицию выполнения; весь стек вызывающей программы не «замораживается»
При следующем next() — продолжает с этой точки

def gen_with_return():
    yield 1
    yield 2
    return "done"          # значение return доступно через StopIteration

g = gen_with_return()
next(g)  # 1
next(g)  # 2
try:
    next(g)
except StopIteration as e:
    print(e.value)  # "done"

#`send()` — двунаправленная коммуникация

def accumulator():
    total = 0
    while True:
        value = yield total   # yield возвращает total И получает значение через send()
        if value is None:
            break
        total += value

gen = accumulator()
next(gen)          # запускаем генератор (до первого yield)
gen.send(10)       # 10
gen.send(20)       # 30
gen.send(5)        # 35

Важно: первый вызов должен быть next(gen) или gen.send(None) — нельзя сразу отправить значение незапущенному генератору.

#`throw()` и `close()`

def safe_gen():
    try:
        while True:
            yield
    except GeneratorExit:
        print("Генератор закрыт")       # вызывается при gen.close()
    except ValueError as e:
        print(f"Получено исключение: {e}")

g = safe_gen()
next(g)
g.throw(ValueError, "ошибка!")   # вбрасывает исключение в точке yield
g.close()                        # завершает генератор (вызывает GeneratorExit)

#3. `yield from` — делегирование

yield from iterable делегирует итерацию другому генератору или итерируемому.

# Без yield from:
def chain_v1(a, b):
    for item in a:
        yield item
    for item in b:
        yield item

# С yield from:
def chain_v2(a, b):
    yield from a
    yield from b

list(chain_v2([1, 2], [3, 4]))  # [1, 2, 3, 4]

Главное преимущество — поддержка send() и throw() сквозь делегирование:

def flatten(nested):
    """Рекурсивное выравнивание вложенных списков"""
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)    # рекурсивное делегирование
        else:
            yield item

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

#4. Генераторные выражения

Синтаксис: (expr for item in iterable if condition) — ленивый аналог list comprehension.

squares = (x**2 for x in range(10))   # генератор, 0 памяти
print(next(squares))   # 0
print(list(squares))   # [1, 4, 9, 16, 25, 36, 49, 64, 81]

# vs
list_sq = [x**2 for x in range(10)]   # список, занимает память сразу

# Использование как аргумента функции — скобки не нужны:
total = sum(x**2 for x in range(10))   # 285

Форма	Тип результата	Память	Многоразовый
`[x for x in ...]`	list	O(n) сразу	да
`(x for x in ...)`	generator	O(1)	нет
`{x for x in ...}`	set	O(n) сразу	да
`{k: v for k, v in ...}`	dict	O(n) сразу	да

#5. Встроенные итераторы Python 3

В Python 3 многие функции возвращают итераторы (в Python 2 они возвращали списки):

m = map(str, [1, 2, 3])        # <map object>
f = filter(None, [0, 1, 2, ""])  # <filter object>
z = zip([1, 2], ['a', 'b'])    # <zip object>
r = reversed([1, 2, 3])        # <list_reverseiterator>

# Нужен список — явный вызов list():
list(map(str, [1, 2, 3]))   # ['1', '2', '3']

Функция	Описание
`range(start, stop, step)`	Ленивая последовательность целых чисел
`enumerate(it, start=0)`	`(index, value)` кортежи
`zip(*iterables)`	Поэлементные кортежи (по shortest)
`map(func, *iterables)`	Применяет func к каждому элементу
`filter(func, iterable)`	Фильтрует элементы
`reversed(seq)`	Обратный итератор (только для последовательностей)

#`any()` и `all()` — короткое замыкание

# any() останавливается при первом True
any(x > 5 for x in range(100))   # True, останавливается на x=6

# all() останавливается при первом False
all(x < 10 for x in [1, 2, 100, 3])  # False, останавливается на 100

# Это делает их эффективными для ленивой проверки условий

#6. `itertools` — операции над итераторами

from itertools import (
    count, cycle, repeat,           # бесконечные
    chain, islice, takewhile,       # конечные
    dropwhile, groupby, accumulate, # конечные
    product, permutations,          # комбинаторные
    combinations, combinations_with_replacement
)

#Бесконечные итераторы

from itertools import count, cycle, repeat

count(10)           # 10, 11, 12, 13, ...
count(0, 0.5)       # 0, 0.5, 1.0, 1.5, ...

cycle([1, 2, 3])    # 1, 2, 3, 1, 2, 3, ...

repeat(42)          # 42, 42, 42, ... (бесконечно)
repeat(42, 3)       # 42, 42, 42 (ровно 3 раза)

# Использование с islice для ограничения:
from itertools import islice
list(islice(count(0), 5))   # [0, 1, 2, 3, 4]

#Конечные итераторы

from itertools import chain, islice, takewhile, dropwhile, groupby, accumulate

# chain — объединение итераторов
list(chain([1, 2], [3, 4], [5]))           # [1, 2, 3, 4, 5]
list(chain.from_iterable([[1, 2], [3, 4]]))  # [1, 2, 3, 4]

# islice — срез итератора
list(islice([1,2,3,4,5], 3))        # [1, 2, 3]
list(islice([1,2,3,4,5], 1, 4))     # [2, 3, 4]
list(islice([1,2,3,4,5], 0, 5, 2))  # [1, 3, 5]

# takewhile — пока предикат истинен
list(takewhile(lambda x: x < 5, [1, 3, 7, 2, 4]))  # [1, 3]

# dropwhile — пропускает пока предикат истинен
list(dropwhile(lambda x: x < 5, [1, 3, 7, 2, 4]))  # [7, 2, 4]

# accumulate — кумулятивные операции
from itertools import accumulate
import operator
list(accumulate([1, 2, 3, 4]))                   # [1, 3, 6, 10] (кумулятивная сумма)
list(accumulate([1, 2, 3, 4], operator.mul))      # [1, 2, 6, 24] (кумулятивное произведение)
list(accumulate([3, 1, 4, 1, 5], max))            # [3, 3, 4, 4, 5] (бегущий максимум)

# groupby — группировка смежных элементов по ключу
from itertools import groupby
data = sorted([("a", 1), ("b", 2), ("a", 3)], key=lambda x: x[0])
for key, group in groupby(data, key=lambda x: x[0]):
    print(key, list(group))
# a [('a', 1), ('a', 3)]
# b [('b', 2)]

Важно: Ловушка groupby: функция группирует только смежные элементы с одинаковым ключом. Сортируйте по тому же ключу, если нужна одна общая группа на ключ, как в SQL. Для run-length encoding или уже упорядоченного потока предварительная сортировка не нужна.

#Комбинаторные итераторы

from itertools import product, permutations, combinations, combinations_with_replacement

# product — декартово произведение (как вложенные циклы for)
list(product([1, 2], ['a', 'b']))    # [(1,'a'), (1,'b'), (2,'a'), (2,'b')]
list(product("AB", repeat=2))       # ('A','A'), ('A','B'), ('B','A'), ('B','B')

# permutations — перестановки
list(permutations([1, 2, 3], 2))    # (1,2), (1,3), (2,1), (2,3), (3,1), (3,2)

# combinations — комбинации без повторений
list(combinations([1, 2, 3], 2))    # (1,2), (1,3), (2,3)

# combinations_with_replacement — с повторениями
list(combinations_with_replacement([1, 2], 2))  # (1,1), (1,2), (2,2)

#7. Паттерны и лучшие практики

#Пайплайн генераторов

def read_lines(filename):
    """Читаем файл строка за строкой — не загружая всё в память"""
    with open(filename) as f:
        yield from f

def parse_lines(lines):
    """Парсим каждую строку"""
    for line in lines:
        yield line.strip().split(',')

def filter_empty(rows):
    """Фильтруем пустые строки"""
    for row in rows:
        if any(row):
            yield row

# Компоновка пайплайна:
lines = read_lines('data.csv')
rows = parse_lines(lines)
valid = filter_empty(rows)
for row in valid:
    process(row)
# Вся цепочка работает лениво — каждый элемент проходит через весь пайплайн

#Ленивые вычисления vs жадные

# Жадно — загружает всё в память
urls = ['...'] * 1_000_000
results = [fetch(url) for url in urls]   # 1M запросов сразу!

# Лениво — генератор, запросы по мере необходимости
results = (fetch(url) for url in urls)
for result in results:
    process(result)   # fetch вызывается только здесь

#Защита от повторного использования итератора

def safe_iter(data):
    """Возвращает итератор; новый — только для повторно итерируемого объекта."""
    return iter(data)

values = [1, 2, 3]
assert list(safe_iter(values)) == [1, 2, 3]
assert list(safe_iter(values)) == [1, 2, 3]  # list создаёт новый iterator

# Антипаттерн: сохранение итератора и попытка пройти дважды
it = iter([1, 2, 3])
first_pass = list(it)    # [1, 2, 3]
second_pass = list(it)   # [] — итератор исчерпан!

Если data уже является итератором, iter(data) is data, поэтому safe_iter не делает его повторно используемым. Для второго прохода храните исходную коллекцию, создавайте iterator фабрикой или осознанно материализуйте данные.

#Числовые итераторы для математики

from itertools import islice

# Простые числа (решето Эратосфена через генераторы)
def primes():
    """Бесконечный генератор простых чисел"""
    sieve = {}
    n = 2
    while True:
        if n not in sieve:
            yield n
            sieve[n * n] = [n]
        else:
            for p in sieve[n]:
                sieve.setdefault(p + n, []).append(p)
            del sieve[n]
        n += 1

first_10 = list(islice(primes(), 10))  # [2, 3, 5, 7, 11, 13, 17, 19, 23, 29]

#8. Асинхронные итераторы и генераторы

Отложенное углубление: этот раздел не нужен для следующей темы. Сначала пройдите async_python, затем вернитесь сюда: async for требует понимания корутин, await и event loop.

import asyncio

# Асинхронный генератор
async def async_range(n):
    for i in range(n):
        await asyncio.sleep(0)   # yield control
        yield i

# Использование через async for:
async def main():
    async for val in async_range(5):
        print(val)

# Асинхронный итератор через класс:
class AsyncCounter:
    def __init__(self, stop):
        self.current = 0
        self.stop = stop

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.current >= self.stop:
            raise StopAsyncIteration
        await asyncio.sleep(0)
        self.current += 1
        return self.current

#9. Типичные ошибки

#Ошибка 1: Забыть что итератор одноразовый

gen = (x**2 for x in range(5))
print(sum(gen))   # 30
print(sum(gen))   # 0 — уже исчерпан!

# Решение: функция-фабрика или list:
def squares(n):
    return (x**2 for x in range(n))

print(sum(squares(5)))   # 30
print(sum(squares(5)))   # 30 — каждый раз новый генератор

#Ошибка 2: Изменение контейнера во время итерации

lst = [1, 2, 3, 4]
for item in lst:
    if item % 2 == 0:
        lst.remove(item)   # RuntimeError или пропуск элементов!

# Решение: итерируйте по копии:
for item in lst[:]:
    if item % 2 == 0:
        lst.remove(item)

#Ошибка 3: ожидать от `groupby` SQL-подобную группировку

data = [('a', 1), ('b', 2), ('a', 3)]   # не отсортированы
for key, group in groupby(data, key=lambda x: x[0]):
    print(key, list(group))
# a [('a', 1)]  ← только первый 'a'!
# b [('b', 2)]
# a [('a', 3)]  ← второй 'a' отдельная группа

# Если нужна одна группа на ключ, сначала сортируем:
for key, group in groupby(sorted(data, key=lambda x: x[0]), key=lambda x: x[0]):
    print(key, list(group))

#Практика: от пробы к рабочему решению

#Шаг 1. Соберите минимальный пример

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

#Шаг 2. Проверьте поведение

from collections.abc import Iterable, Iterator
from itertools import islice


def batches(items: Iterable[int], size: int) -> Iterator[tuple[int, ...]]:
    if size < 1:
        raise ValueError("size должен быть положительным")
    iterator = iter(items)
    while batch := tuple(islice(iterator, size)):
        yield batch


assert list(batches((value for value in range(7)), 3)) == [
    (0, 1, 2),
    (3, 4, 5),
    (6,),
]

#Шаг 3. Доведите решение до рабочего сценария

Практика с подсказками: обработайте большой текстовый файл пакетами. Внесите сбой на середине чтения, докажите закрытие файла и зафиксируйте верхнюю границу размера пакета. Замена списка на yield без такого ограничения не считается доказательством ограниченной памяти.

#Шаг 4. Объясните внутренний механизм

Исследуйте send, throw и close на генераторе с try/finally. Запишите трассу состояний и объясните, кто владеет закрытием при раннем завершении потребителя.

#Упражнения

Реализуйте итератор InfiniteCounter(start, step) через класс.
Напишите генераторный пайплайн для чтения CSV без загрузки в память.
Используя itertools.groupby, сгруппируйте слова по первой букве.
Реализуйте take(n, iterable) через islice.
Объясните, почему any() и all() эффективны с генераторами.

#Самопроверка

iter(counter) is counter, а последовательные next() изменяют состояние ровно на step; явно решите, допустим ли отрицательный шаг.
CSV-пайплайн читает лениво: получение первых двух записей не должно читать весь файл, а файл закрывается при ошибке потребителя.
Перед groupby данные отсортированы по тому же ключу; иначе одинаковые буквы могут образовать несколько групп.
take(0, ...) возвращает пустой результат и не потребляет исходный iterator.
Докажите короткое замыкание any/all генератором со счётчиком вычисленных элементов.

#Практическая лаборатория

Закрепите тему в лаборатории «Ленивый поток JSONL с ограниченной памятью». Вы проверите ленивость по числу чтений, обработаете бесконечный источник и зафиксируете контракт закрытия генератора при ранней остановке.

git clone --branch v0.6.2 --depth 1 https://gitlab.potapov.me/courses/python-labs.git
cd python-labs
uv sync --group test
uv run --group test pytest iterators_generators/tests

Форма

Тип результата

Память

Многоразовый

[x for x in ...]

list

O(n) сразу

да

(x for x in ...)

generator

O(1)

нет

{x for x in ...}

set

O(n) сразу

да

{k: v for k, v in ...}

dict

O(n) сразу

да

Функция

Описание

range(start, stop, step)

Ленивая последовательность целых чисел

enumerate(it, start=0)

(index, value) кортежи

zip(*iterables)

Поэлементные кортежи (по shortest)

map(func, *iterables)

Применяет func к каждому элементу

filter(func, iterable)

Фильтрует элементы

reversed(seq)

Обратный итератор (только для последовательностей)