Итераторы, генераторы и корутины

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

Эти механизмы позволяют обрабатывать файлы, не помещающиеся в память, и описывать бесконечные последовательности. Из генераторов выросла и асинхронность Python: первые корутины asyncio были генераторами, а нативные корутины async def используют тот же механизм приостановки функции. Сам asyncio рассматривается в главе, посвящённой асинхронности.

Слайды к главе. Материал главы изложен также в лекции «Итераторы, генераторы и корутины» (итераторы, генераторы, модуль itertools, сопрограммы, async и await) с демонстрациями в интерактивной оболочке; слайды лекции доступны на сайте книги и в PDF.

Итераторы

Устройство цикла for

Цикл for в Python не перебирает индексы: он получает у объекта итератор функцией iter и вызывает для него next, пока тот не сообщит об окончании исключением StopIteration. Исключение перехватывает сам цикл, поэтому в коде оно не видно. Тот же цикл, записанный явно:

xs = [1.5, 2.5, 4.0]

it = iter(xs)
while True:
    try:
        x = next(it)
    except StopIteration:
        break
    print(x)

Встроенные функции iter и next вызывают специальные методы __iter__ и __next__, как len вызывает __len__. По тому же протоколу работают распаковка, списковые включения, функции sum, max, sorted и list и конструкторы коллекций. В байт-коде цикл представлен двумя инструкциями: GET_ITER вызывает iter для перебираемого объекта, FOR_ITER на каждом шаге вызывает next и при окончании элементов переходит за конец цикла.

import dis

dis.dis('for x in xs: pass')

  0           0 RESUME                   0

  1           2 LOAD_NAME                0 (xs)
              4 GET_ITER
        >>    6 FOR_ITER                 2 (to 14)
             10 STORE_NAME               1 (x)
             12 JUMP_BACKWARD            4 (to 6)
        >>   14 END_FOR
             16 RETURN_CONST             0 (None)

Вывод получен в Python 3.12; набор инструкций меняется от версии к версии, но пара GET_ITER и FOR_ITER присутствует во всех.

Итерируемый объект и итератор

Протокол итерации разделяет две роли. Итерируемый объект (iterable) обладает методом __iter__, который при каждом вызове возвращает новый итератор. Итератор (iterator) обладает методом __next__, возвращающим следующий элемент или возбуждающим StopIteration, когда элементы закончились, и методом __iter__, возвращающим сам итератор. Итератор обязан быть и итерируемым, поэтому его можно передать в цикл for.

Встроенные коллекции — списки, кортежи, словари, множества, строки — являются итерируемыми объектами; файл, открытый функцией open, сам является итератором по строкам. По списку можно пройти сколько угодно раз, поскольку каждый цикл получает новый итератор, а по итератору — один раз: после первого прохода он исчерпан. Исчерпанный итератор обязан и дальше возбуждать StopIteration при каждом вызове next.

xs = [1.5, 2.5, 4.0]
it = iter(xs)
next(it), next(it)    # (1.5, 2.5)
next(it)              # 4.0
next(it)              # исключение StopIteration
iter(it) is it        # True: итератор возвращает самого себя
list(it), list(xs)    # ([], [1.5, 2.5, 4.0])
next(range(3))        # TypeError: 'range' object is not an iterator

Проверка isinstance(x, collections.abc.Iterator) отличает итератор от итерируемого объекта, у которого есть только __iter__.

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

Класс Linspace ниже — упрощённый аналог numpy.linspace: n равноотстоящих точек отрезка от a до b. Роли разделены явно: контейнер хранит параметры сетки и на каждый вызов __iter__ создаёт новый итератор, а обход выполняет класс LinspaceIterator.

class Linspace:
    def __init__(self, a, b, n):
        self.a, self.n = a, n
        self.h = (b - a) / (n - 1)

    def __iter__(self):
        return LinspaceIterator(self.a, self.h, self.n)

class LinspaceIterator:
    def __init__(self, a, h, n):
        self.a, self.h, self.n, self.i = a, h, n, 0   # состояние обхода

    def __iter__(self):
        return self                # итератор обязан быть итерируемым

    def __next__(self):
        if self.i == self.n:
            raise StopIteration
        self.i += 1
        return self.a + (self.i - 1) * self.h

grid = Linspace(0.0, 1.0, 5)
list(grid)    # [0.0, 0.25, 0.5, 0.75, 1.0]
list(grid)    # тот же результат: второй проход получает новый итератор

Счётчик i хранится в итераторе, а не в контейнере, поэтому каждый новый for начинает с первой точки, а по одной сетке можно пройти произвольное число раз. Точка вычисляется как a + i·h, а не прибавлением h к предыдущей: при накоплении ошибки округления складываются и растут с номером точки, а при умножении погрешность каждой точки остаётся в последних разрядах. Последняя точка и при умножении может отличаться от b в последнем знаке, поэтому numpy.linspace присваивает ей значение b явно.

Обе роли можно совместить в одном классе, где __iter__ возвращает self: кода при этом меньше, однако второй проход начнётся там, где закончился первый, и окажется пустым. Так устроены однопроходные объекты — файлы и генераторы; у контейнеров роли разделены. Без метода __iter__ сам итератор не является итерируемым: цикл for по контейнеру Linspace работает, поскольку у полученного итератора цикл вызывает только __next__, а цикл по объекту iter(grid) и передача его в list или zip завершаются TypeError.

Другие способы итерации

Кроме основного протокола, функция iter поддерживает ещё два способа получить итератор, а оператор in — свой порядок поиска.

Если у объекта нет __iter__, но есть __getitem__, функция iter создаёт итератор, запрашивающий элементы с индексами 0, 1, 2 и далее, пока __getitem__ не возбудит IndexError. Так перебирались последовательности до появления протокола итераторов, и этот путь сохранён для совместимости. collections.abc.Iterable такой объект итерируемым не признаёт, поэтому надёжная проверка — вызов iter(obj).

class Squares:
    def __getitem__(self, i):
        if i >= 4:
            raise IndexError(i)
        return i * i

list(Squares()), 9 in Squares()   # ([0, 1, 4, 9], True)

Функция iter с двумя аргументами принимает вызываемый объект и граничное значение: итератор вызывает функцию без аргументов и выдаёт результаты, пока не получит граничное значение. Двоичные файлы оцифровщиков и детекторов читаются так блоками фиксированного размера; границей служит пустая строка байтов, которую read возвращает в конце файла.

from functools import partial

with open('run042.dat', 'rb') as f:
    for block in iter(partial(f.read, 4096), b''):
        process(block)

Оператор in вызывает метод __contains__, а при его отсутствии перебирает объект через __iter__ или __getitem__, пока не найдёт равный элемент. Для итератора это означает расход элементов: после проверки в нём остаются только элементы после найденного.

it = iter(Linspace(0.0, 1.0, 5))
0.5 in it      # True
list(it)       # [0.75, 1.0]: элементы до 0.5 включительно израсходованы

Итераторы стандартной библиотеки

range не хранит элементы, а вычисляет их и при этом остаётся последовательностью с длиной, индексами и проверкой in без перебора для целых чисел. Представления словаря отражают его текущее содержимое; изменение размера словаря во время обхода завершается RuntimeError: dictionary changed size during iteration. Функции map, filter, zip, enumerate и reversed возвращают итераторы и вычисляют элементы по запросу, поэтому их результат, использованный дважды, во второй раз пуст — частая ошибка при переносе кода с Python 2, где map и zip возвращали списки.

m = map(str, [1, 2])
list(m), list(m)                      # (['1', '2'], [])
list(zip([1, 2], [1], strict=True))   # ValueError: zip() argument 2 is shorter than argument 1

Функция zip останавливается на самой короткой последовательности; с Python 3.10 параметр strict=True превращает расхождение длин в ValueError. При объединении отметок времени и показаний zip(times, values, strict=True) защищает от молчаливого усечения рядов разной длины.

Генераторы

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

Писать два класса ради простого перебора избыточно. Функция, в теле которой встречается yield, при вызове не выполняет тело, а возвращает объект-генератор. Генератор — итератор: next запускает тело и выполняет его до ближайшего yield, значение yield становится результатом next, а функция приостанавливается. Генератор хранит кадр функции — локальные переменные и место остановки — между вызовами next; точку остановки задаёт yield. Следующий next продолжает выполнение с инструкции после yield.

def countdown(n):
    print('старт')
    while n > 0:
        yield n
        n -= 1
    return 'пуск'

g = countdown(3)    # тело не выполняется, ничего не печатается
next(g)             # печатает «старт», возвращает 3
next(g), next(g)    # (2, 1)
next(g)             # исключение StopIteration: пуск

Когда тело завершается, возбуждается StopIteration, и значение return, если оно есть, сохраняется в атрибуте value исключения; цикл for и функция list это значение не используют: list(countdown(3)) печатает «старт» и возвращает [3, 2, 1]. Генератор заменяет класс итератора: метод __iter__, написанный генератором, возвращает новый генератор при каждом вызове, и по контейнеру можно проходить многократно без отдельного класса.

class Linspace:
    def __init__(self, a, b, n):
        self.a, self.n = a, n
        self.h = (b - a) / (n - 1)

    def __iter__(self):
        for i in range(self.n):
            yield self.a + i * self.h

Генератор удобен там, где следующий элемент вычисляется по предыдущим, — шаги интегрирования, итерации метода Ньютона, последовательные приближения.

Состояния генератора

Генератор проходит четыре состояния, которые возвращает функция inspect.getgeneratorstate.

import inspect

g = countdown(3)
inspect.getgeneratorstate(g)                       # 'GEN_CREATED'
next(g)                                            # печатает «старт», возвращает 3
inspect.getgeneratorstate(g), g.gi_frame.f_locals  # ('GEN_SUSPENDED', {'n': 3})

Состояние хранится в атрибутах объекта: gi_frame — кадр выполнения с локальными переменными f_locals и номером строки f_lineno, после завершения генератора gi_frame равен None; gi_running — признак выполнения; gi_yieldfrom — подгенератор, которому делегировано выполнение через yield from. Методы send, throw и close рассматриваются в разделе о корутинах; close у ещё не запущенного генератора переводит его в GEN_CLOSED сразу, без выполнения тела. Генератор не может запустить сам себя: next(g) внутри тела g завершается ValueError: generator already executing.

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

Генераторное выражение — краткая форма генератора; от спискового включения его отличают круглые скобки вместо квадратных. Если выражение — единственный аргумент функции, вторые скобки не нужны.

squares = [x * x for x in range(100_000)]   # список строится сразу
lazy = (x * x for x in range(100_000))      # генератор, элементы по запросу
total = sum(x * x for x in range(100_000))  # скобки вызова служат скобками выражения

Списковое включение хранит все элементы, генератор не вычисляет ни одного, пока элемент не будет запрошен. Измерения на Python 3.12.3:

sys.getsizeof учитывает только сам список, то есть массив ссылок; объекты-числа добавляют ещё около 2,8 МБ, что и показывает пик памяти при суммировании. Размер генератора не зависит от длины последовательности, но зависит от версии Python: в 3.11 тот же генератор занимает 208 байт.

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

data = [0, 2, 5, 12]
threshold = 1
g = (x for x in data if x > threshold)
threshold = 10
list(g)          # [12]: условие прочитало threshold в момент перебора

Суммы, средние и экстремумы по большим наборам удобно вычислять генераторными выражениями — без промежуточного списка в памяти.

Конвейер генераторов

Генераторы соединяются в конвейер: каждая стадия, кроме первой, принимает итерируемый объект и сама является генератором. Журнал захода, записывавшийся всю смену, может не помещаться в память целиком, но построчно читается так же просто, как список.

def read_lines(path):
    with open(path) as f:
        yield from f

def parse(lines):
    for line in lines:
        if not line.strip() or line.startswith('#'):
            continue           # пустые строки и комментарии пропускаются
        t, channel, value = line.split()
        yield float(t), channel, float(value)

def select(events, name):
    for t, channel, value in events:
        if channel == name:
            yield t, value

lines = read_lines('run042.log')
bpm = select(parse(lines), 'BPM01')
print(max(value for t, value in bpm))

read_lines читает журнал построчно, поскольку файл — итератор по строкам; parse пропускает пустые строки и комментарии и выдаёт разобранные события; select оставляет события одного канала; max потребляет результат. Элементы проходят конвейер по одному: max запрашивает событие у select, select — у parse, parse — строку у read_lines. В памяти одновременно находятся одна строка и одно событие, поэтому журнал любого размера обрабатывается с постоянным расходом памяти. Стадии независимы: их легко заменить, добавить проверку или пересчёт единиц, а parse и select проверяются по отдельности на списке строк вместо файла.

log = ['0.0 BPM01 1.2', '0.1 BPM02 0.7', '0.2 BPM01 1.4']
pipe = select(parse(log), 'BPM01')   # ничего не прочитано
next(pipe)                           # (0.0, 1.2)
next(pipe)                           # (0.2, 1.4): строка BPM02 отброшена
next(pipe, 'конец')                  # 'конец' вместо StopIteration

Файл закрывается, когда генератор read_lines завершится или будет закрыт; если обход прерван досрочно, файл остаётся открытым до вызова close или сборки мусора (раздел «Завершение генератора»).

Бесконечные последовательности

Генератор с бесконечным циклом описывает последовательность без конца: значения вычисляются, пока их запрашивают, а границу устанавливает читающая сторона. Генератор oscillator выдаёт состояния гармонического осциллятора с шагом dt по полунеявной схеме Эйлера: сначала обновляется скорость, затем координата по новой скорости.

import itertools, math

def oscillator(x, v, dt, w=1.0):
    t = 0.0
    while True:
        yield t, x, v
        v -= w * w * x * dt
        x += v * dt
        t += dt

states = oscillator(1.0, 0.0, 0.01)
list(itertools.islice(states, 3))   # три первых состояния
next(states)                        # четвёртое, t = 0.03: генератор продолжает с места остановки
next(t for t, x, v in oscillator(1.0, 0.0, 0.01) if x < 0)   # 1.5700000000000012
math.pi / 2                                                  # 1.5707963267948966

Условия остановки в генераторе нет: его задаёт вызывающий код. itertools.islice берёт заданное число состояний и больше не обращается к генератору, takewhile выдаёт состояния, пока выполняется условие, next с генераторным выражением возвращает первое состояние, удовлетворяющее условию, например первое пересечение нуля. takewhile при этом забирает из генератора и первый элемент, не прошедший условие, и этот элемент теряется. Список из бесконечного генератора построить нельзя: list(oscillator(...)) не завершится, а память будет расти, пока процесс не будет остановлен.

Координата впервые становится отрицательной при t = 1,57 при точном значении π/2 = 1,5708: расхождение меньше шага. Полунеявная схема точно сохраняет величину x² + v² − dt·x·v, поэтому x² + v² колеблется в узкой полосе около единицы: за 10 000 шагов отклонение не превышает dt/(2 − dt) ≈ 0,00503. У явной схемы Эйлера x² + v² на каждом шаге умножается на 1 + dt², и за те же 10 000 шагов выросла бы примерно в e раз. Так генератор разделяет схему интегрирования и условие остановки — по времени, по числу шагов или по событию.

Моменты времени, вычисляемые сложением t += dt, тоже накапливают ошибку округления: у itertools.count(0.0, 0.1) десятый шаг даёт 0.9999999999999999 вместо 1.0, и документация itertools рекомендует в таком случае умножение. Генератор моментов времени без накопления:

def clock(dt):
    """Бесконечная последовательность моментов времени с шагом dt."""
    for i in itertools.count():
        yield i * dt           # умножение, а не накопление t += dt

[round(t, 3) for t in itertools.islice(clock(0.02), 5)]   # [0.0, 0.02, 0.04, 0.06, 0.08]

Делегирование: yield from

Когда генератор лишь передаёт элементы другого итерируемого объекта, цикл for x in g: yield x записывается короче: yield from g. Рекурсивный обход двоичного дерева поиска в порядке возрастания выдаёт элементы левого поддерева, значение узла и элементы правого поддерева, а вложенные вызовы соединяются через yield from.

class Node:
    def __init__(self, value, left=None, right=None):
        self.value = value
        self.left, self.right = left, right

def walk(node):
    if node is not None:
        yield from walk(node.left)
        yield node.value
        yield from walk(node.right)

tree = Node(5, Node(3, Node(1)), Node(8))
list(walk(tree))     # [1, 3, 5, 8]

Конструкция yield from (PEP 380, Python 3.3) не сводится к сокращению цикла. Она передаёт во вложенный генератор значения, отправленные через send, и исключения, отправленные через throw, а значением самого выражения yield from становится результат return вложенного генератора:

def sub():
    yield 1
    yield 2
    return 'итог'

def outer():
    r = yield from sub()
    print('подгенератор вернул', r)

list(outer())   # печатает «подгенератор вернул итог», результат [1, 2]

Это позволяет разбивать сложный генератор на части, как функцию на подфункции. Рекурсивным генератором с yield from записывается обход любых деревьев — разбиений пространства (квадродеревья и октодеревья в задаче N тел), иерархий геометрии детектора.

Завершение генератора

До Python 3.7 исключение StopIteration, возникшее внутри тела генератора, — например, от вызова next у исчерпанного итератора, — молча завершало генератор, и ошибка терялась. PEP 479 изменил это поведение: такое исключение превращается в RuntimeError. Поэтому внутри генератора next вызывается со значением по умолчанию или в блоке try, а для завершения используется return.

def firsts(groups):
    for g in groups:
        yield next(iter(g))        # пустая группа: RuntimeError

list(firsts([[1], []]))   # RuntimeError: generator raised StopIteration

def firsts(groups):
    for g in groups:
        yield next(iter(g), None)

list(firsts([[1], []]))   # [1, None]

Генератор, обход которого прерван досрочно — оператором break или исключением у потребителя, — остаётся в состоянии GEN_SUSPENDED; его блоки finally и менеджеры контекста выполнятся при вызове close или при уничтожении объекта сборщиком мусора. В CPython это обычно происходит сразу после потери последней ссылки, в других реализациях — позже, поэтому ресурсы освобождают явно: g.close() или contextlib.closing(g).

Модуль itertools

Модуль itertools содержит ленивые итераторы, реализованные на C, — базовые средства обработки последовательностей без промежуточных списков. Все функции модуля возвращают итераторы, поэтому их результат однопроходный и может быть входом следующей функции.

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

count считает от start с шагом step, в том числе для вещественных чисел, но с вещественным шагом накапливает его сложением; cycle повторяет элементы источника по кругу, сохраняя их копию; repeat выдаёт одно значение заданное или бесконечное число раз. islice — аналог среза для любого итератора без отрицательных индексов, поскольку длина итератора неизвестна. takewhile останавливается на первом ложном условии, dropwhile пропускает элементы, пока условие истинно, а после первого неподходящего выдаёт всё остальное без проверки: это не фильтр.

Объединение и группировка

chain.from_iterable соединяет источники из итерируемого объекта, например из списка списков; распаковка chain(*xss) строит кортеж источников и на бесконечном источнике не завершится. groupby выдаёт пары «ключ, итератор группы» и объединяет только соседние элементы, поэтому для группировки по всему набору данные сначала сортируют по тому же ключу.

from itertools import groupby, pairwise, batched, accumulate

events = [('BPM02', 0.7), ('BPM01', 1.2), ('BPM01', 1.4), ('BPM02', 0.9)]
[name for name, _ in groupby(events, key=lambda e: e[0])]
# ['BPM02', 'BPM01', 'BPM02']: без сортировки BPM02 попал в две группы
for name, grp in groupby(sorted(events), key=lambda e: e[0]):
    print(name, [v for _, v in grp])   # BPM01 [1.2, 1.4] и BPM02 [0.7, 0.9]

times = [0.00, 0.12, 0.31, 0.39]
[round(b - a, 2) for a, b in pairwise(times)]   # [0.12, 0.19, 0.08]
list(batched(range(7), 3))                      # [(0, 1, 2), (3, 4, 5), (6,)]
list(accumulate([3, 1, 4, 1, 5], max))          # [3, 3, 4, 4, 5]

pairwise по отметкам времени даёт интервалы между событиями, accumulate — накопленную дозу или интеграл методом прямоугольников, batched — пакеты для записи в файл или отправки по сети. Вместо pairwise иногда пишут zip(xs, islice(xs, 1, None)); на списке это работает, а на итераторе даёт неверные пары, поскольку оба аргумента расходуют один итератор.

Комбинаторика

Комбинаторные функции заменяют вложенные циклы: product выдаёт декартово произведение в лексикографическом порядке, где последний источник меняется быстрее всех. Число вариантов растёт быстро, поэтому его заранее оценивают функциями math.perm и math.comb (Python 3.8).

from itertools import product, combinations
import math

grid = list(product([1.0, 2.0], [0.1, 0.2, 0.3]))
len(grid), grid[:3]          # (6, [(1.0, 0.1), (1.0, 0.2), (1.0, 0.3)])
list(combinations('ABCD', 2))
# [('A', 'B'), ('A', 'C'), ('A', 'D'), ('B', 'C'), ('B', 'D'), ('C', 'D')]
math.comb(32, 2), sum(1 for _ in combinations(range(32), 2))   # (496, 496)

product перебирает сетку параметров моделирования — температуры и поля, энергии и углы — одним циклом без вложенности; combinations даёт пары детекторов для поиска совпадений: 32 детектора образуют 496 пар. Число повторов в product передаётся только именованным аргументом repeat: вызов product('ab', 2) завершается TypeError.

Рецепты и ограничения

Документация модуля содержит раздел рецептов — коротких функций, собранных из стандартных блоков: скользящее окно sliding_window, чередование источников roundrobin, удаление повторов unique_everseen и другие; многие из них вошли в сторонний пакет more-itertools. Скользящее среднее строится на deque ограниченной длины: очередь хранит последние n отсчётов, а старые вытесняются автоматически.

from collections import deque
from itertools import islice

def moving_average(xs, n):
    it = iter(xs)
    window = deque(islice(it, n - 1), maxlen=n)
    for x in it:
        window.append(x)
        yield sum(window) / n

list(moving_average([1, 2, 3, 4, 5], 3))   # [2.0, 3.0, 4.0]

Ограничения модуля следуют из ленивости:

• product, permutations и combinations читают входные источники целиком и сохраняют их в кортежах ещё до выдачи первого результата, поэтому бесконечный итератор передавать в них нельзя;
• tee хранит элементы, пока их не прочитают все копии, и при большом отставании одной копии расходует память как список; исходный итератор после tee не используется;
• группа из groupby действительна только до перехода к следующей: если её нужно сохранить, её сразу превращают в список;
• islice не поддерживает отрицательные индексы.

Скользящее среднее и медиана — простейшие фильтры сигнала; для массивов NumPy среднее вычисляют свёрткой numpy.convolve, медиану — функцией scipy.signal.medfilt, а генератор подходит для потока, который не хранится целиком.

Корутины

Корутины (сопрограммы) выросли из генераторов: генератор, принимающий значения через send, хранит состояние между вызовами и обрабатывает поток входных данных. На том же механизме приостановки построена асинхронность, где один поток занимается другой работой, пока задача ожидает ответа от сети или прибора.

Метод send

Выражение yield справа от присваивания делает генератор двусторонним: x = yield result выдаёт result наружу, приостанавливает функцию и при возобновлении получает значение, переданное методом send. Вызов next(g) равносилен g.send(None). Перед первым send корутину запускают вызовом next, чтобы она дошла до первого yield: отправить значение в только что созданный генератор нельзя. Метод send возвращает следующее значение, выданное корутиной, поэтому один вызов и передаёт измерение, и возвращает обновлённый результат.

Корутина stats вычисляет среднее и дисперсию потока измерений по формулам Уэлфорда: на каждом шаге обновляются число отсчётов, среднее и сумма квадратов отклонений M2, а несмещённая оценка дисперсии равна M2/(n − 1).

def stats():
    n, mean, m2 = 0, 0.0, 0.0
    result = None
    while True:
        x = yield result
        n += 1
        d = x - mean
        mean += d / n
        m2 += d * (x - mean)
        result = n, mean, m2 / (n - 1) if n > 1 else 0.0

s = stats()
s.send(2.0)    # TypeError: can't send non-None value to a just-started generator
next(s)        # запуск до первого yield
s.send(2.0)    # (1, 2.0, 0.0)
s.send(4.0)    # (2, 3.0, 2.0)
s.send(6.0)    # (3, 4.0, 4.0)

Состояние хранится в локальных переменных генератора, а не в атрибутах объекта; эквивалентный класс с методом add длиннее, но при сложном состоянии нагляднее. Формулы Уэлфорда численно устойчивы: в отличие от формулы «среднее квадратов минус квадрат среднего», они не вычитают близкие большие числа и не теряют точность при большом постоянном смещении сигнала.

Тот же приём с аннотацией типов: тип Generator[тип выдачи, тип приёма, тип возврата] из collections.abc описывает корутину, выдающую None или результат, принимающую вклад и ничего не возвращающую. Корутина ниже вычисляет вклад с учётом сложных процентов; коэффициент роста вычисляется один раз — при первом next(coro), когда корутина доходит до первого yield, — и далее не меняется, а суммы вклада в неё можно передавать неограниченно.

from collections.abc import Generator

def cash_return_coro(percent: float,
                     years: int) -> Generator[float | None, float, None]:
    value = (1 + percent / 100) ** years
    result = None
    while True:
        try:
            deposit = yield result
        except GeneratorExit:
            print('Выход из корутины')
            raise
        result = round(deposit * value, 2)

coro = cash_return_coro(5, 5)
next(coro)                       # запуск до первого yield
for item in [1000, 2000, 5000, 10000, 100000]:
    print(coro.send(item))       # 1276.28, 2552.56, 6381.41, 12762.82, 127628.16
coro.close()                     # печатает «Выход из корутины»

Выражение deposit = yield result отдаёт наружу результат предыдущего шага и принимает следующий вклад, поэтому next(coro) нужен один раз — чтобы довести корутину до первого yield. Форма с двумя yield за оборот — один принимает вклад, другой отдаёт результат — требует вызова next после каждого send: если его пропустить, следующий вклад попадёт не в тот yield и потеряется без ошибки.

Методы throw и close

throw возбуждает исключение в точке, где генератор приостановлен, — так, будто его выбросила инструкция yield. Если генератор перехватывает исключение, он продолжает работу до следующего yield, и throw возвращает выданное значение; если нет, исключение выходит наружу из вызова throw. В счётчике ниже ValueError служит командой сброса.

def counter():
    n = 0
    try:
        while True:
            try:
                yield n
                n += 1
            except ValueError:
                n = 0
    finally:
        print('закрытие, n =', n)

c = counter()
next(c), next(c), next(c)   # (0, 1, 2)
c.throw(ValueError)         # 0: сброс без завершения генератора
next(c)                     # 1
c.close()                   # печатает «закрытие, n = 1»
next(c)                     # исключение StopIteration

close возбуждает в точке остановки GeneratorExit: генератор обязан завершиться — выполнить блоки finally и выйти. Если вместо этого он снова выполнит yield, возникнет RuntimeError: generator ignored GeneratorExit. Поэтому после перехвата GeneratorExit генератор не выполняет yield: исключение возбуждают повторно, как в cash_return_coro, или завершают генератор оператором return, а для освобождения ресурсов достаточно finally или with. CPython вызывает close и при уничтожении незавершённого генератора, поэтому файлы и соединения, открытые в теле генератора, закрываются и при прерванном обходе. Форма throw с тремя аргументами — тип, значение, трассировка — с Python 3.12 устарела: передаётся один аргумент — экземпляр или класс исключения.

yield from как канал

yield from соединяет вызывающий код с подгенератором напрямую: значения, выдаваемые подгенератором, проходят наружу, значения send и исключения throw — внутрь, а делегирующий генератор на это время приостановлен и лишь получает итоговый результат.

def averager():
    total, n = 0.0, 0
    while True:
        x = yield
        if x is None:
            return total / n
        total, n = total + x, n + 1

def collect(results, name):
    while True:
        results[name] = yield from averager()

results = {}
c = collect(results, 'BPM01')
next(c)
for x in (1.25, 1.5, 1.75):
    c.send(x)
c.send(None)       # конец серии
results            # {'BPM01': 1.5}

Корутина averager принимает значения одной серии и по сигналу конца серии — значению None — возвращает среднее; делегирующий генератор collect записывает этот результат в словарь и начинает следующую серию. Для вызывающего кода два уровня неразличимы: он отправляет значения в collect, а они попадают в averager. Без yield from пришлось бы вручную перехватывать StopIteration, передавать send и throw и обрабатывать close. На такой передаче управления по цепочке вложенных генераторов была построена первая версия asyncio: корутины ожидали друг друга через yield from, пока в Python 3.5 не появился await.

Генератор как менеджер контекста

Декоратор contextlib.contextmanager превращает генератор с одним yield в менеджер контекста. Код до yield выполняется при входе в блок with и играет роль __enter__; значение yield связывается с переменной после as; код после yield выполняется при выходе из блока и играет роль __exit__.

import contextlib, time

@contextlib.contextmanager
def timer(label):
    t0 = time.perf_counter()
    try:
        yield
    finally:
        dt = time.perf_counter() - t0
        print(f'{label}: {dt:.2f} с')

with timer('сумма'):
    s = sum(i * i for i in range(10**6))   # печатает «сумма: 0.04 с»; время зависит от машины

with timer('деление'):
    1 / 0      # печатает «деление: 0.00 с», затем ZeroDivisionError выходит наружу

Если в блоке with возникает исключение, оно возбуждается в генераторе в точке yield, поэтому завершающий код помещается в finally — иначе при ошибке он не выполнится. Исключение, перехваченное в генераторе и не возбуждённое повторно, считается обработанным. Такой менеджер короче класса с методами __enter__ и __exit__ и подходит для однократных действий вокруг блока: замер времени этапов обработки, временная смена каталога или настроек, захват и освобождение ресурса. Менеджер из contextmanager применяется и как декоратор функции. В contextlib есть и готовые менеджеры: suppress подавляет указанные исключения, chdir (Python 3.11) временно меняет рабочий каталог, closing вызывает close по выходе, ExitStack собирает переменное число менеджеров; временный каталог создаёт tempfile.TemporaryDirectory.

От генераторов к async и await

Цикл событий на генераторах

Планировщик, переключающийся между задачами, записывается в десяток строк, если задачи — генераторы. Задача выполняет шаг работы и отдаёт управление инструкцией yield. Планировщик хранит очередь готовых задач, берёт первую, продвигает её вызовом next до следующего yield и ставит в конец очереди; завершившаяся задача возбуждает StopIteration и из очереди удаляется.

from collections import deque

def task(name, n):
    for i in range(n):
        print(name, 'шаг', i)
        yield

def run(tasks):
    queue = deque(tasks)
    while queue:
        t = queue.popleft()
        try:
            next(t)
        except StopIteration:
            continue
        queue.append(t)

run([task('прибор A', 3), task('прибор B', 2)])

прибор A шаг 0
прибор B шаг 0
прибор A шаг 1
прибор B шаг 1
прибор A шаг 2

Две задачи чередуются, хотя поток один: переключение происходит только в точках yield, поэтому такая многозадачность называется кооперативной — задача, не дошедшая до yield, удерживает поток. В цикле событий asyncio задача, ожидающая данных, не возвращается в очередь сразу: она регистрирует ожидание у селектора операционной системы и снова становится готовой, когда данные пришли или истёк тайм-аут. Полная реализация такого планировщика разобрана в главе «Асинхронность».

async def и await

До Python 3.5 корутины asyncio были генераторами: они ожидали друг друга через yield from и помечались декоратором @asyncio.coroutine, удалённым в Python 3.11. PEP 492 ввёл нативные корутины: функция, объявленная через async def, при вызове возвращает объект типа coroutine, а await приостанавливает её до завершения ожидаемого объекта — другой корутины, задачи или Future. await допустим только внутри async def, а объект-корутину нельзя перебрать циклом for.

Ожидание приборов с разной задержкой показывает разницу между ожиданием и блокировкой:

import asyncio, time

async def poll(name, delay):
    await asyncio.sleep(delay)      # точка ожидания: управление у цикла событий
    return name, delay

async def blocking(name, delay):
    time.sleep(delay)               # блокирующий вызов: цикл событий стоит
    return name, delay

async def main(f):
    t0 = time.perf_counter()
    res = await asyncio.gather(f('A', 0.3), f('B', 0.5), f('C', 0.8))
    print(res, round(time.perf_counter() - t0, 2))

asyncio.run(main(poll))       # [('A', 0.3), ('B', 0.5), ('C', 0.8)] 0.8
asyncio.run(main(blocking))   # [('A', 0.3), ('B', 0.5), ('C', 0.8)] 1.61

С await задержки перекрываются, и общее время равно наибольшей — 0,8 с; time.sleep не отдаёт управление циклу событий, и опросы выполняются по очереди за 1,6 с. Вызов poll('A', 0.3) без await ничего не выполняет: создаётся объект, а при его уничтожении выводится предупреждение RuntimeWarning: coroutine 'poll' was never awaited — признак забытого await. Задачи, тайм-ауты, группы задач и асинхронная итерация рассматриваются в главе «Асинхронность».

Выбор средства

Итоги

Таким образом, цикл for получает итератор функцией iter и вызывает next, пока тот не возбудит StopIteration; итерируемый объект выдаёт новый итератор на каждый проход, а сам итератор однопроходный. Функция с yield возвращает генератор, выполняемый по шагам: между вызовами локальные переменные и место остановки хранятся в кадре, а return завершает генератор и передаёт значение в StopIteration. Генераторные выражения и конвейеры генераторов обрабатывают данные по одному элементу, поэтому расход памяти не зависит от объёма данных. Модуль itertools даёт ленивые бесконечные итераторы, срезы, объединение, группировку и комбинаторику. Методы send, throw и close превращают генератор в корутину, а yield from соединяет её с подгенератором и возвращает его результат. Нативные корутины async def и await выполняет цикл событий, переключая задачи в точках ожидания.

Для самостоятельного изучения рекомендуются документация модуля itertools с разделом рецептов, PEP 380 (yield from), PEP 479 (StopIteration внутри генератора) и PEP 492 (async и await).