Асинхронные итераторы: async for

Списки и файлы уже ждут тебя целиком: данные где-то лежат, цикл просто перебирает. А в асинхронном мире половина данных так не живёт — события бота, показания датчика, сообщения вебсокета приходят тогда, когда приходят: сейчас три штуки за секунду, потом тишина на десять секунд. Обычный for бессилен — ему нужен готовый список. Для потоков, которые наливаются со временем, у asyncio есть собственный протокол итерации и собственный цикл: async for.

Идея урока: ты научишься писать объект, по которому async for ходит шагами, — и каждый шаг умеет ждать. Телеметрия по тикам, лента событий, очередь сообщений — всё это один и тот же приём: отдал следующий элемент, когда он появился, и не раньше.

Протокол async-итерации: __aiter__ и __anext__

Вспомни, как устроен обычный for: он вызывает у объекта __iter__, чтобы получить итератор, а на каждом витке — __next__, чтобы получить следующий элемент; когда данные кончаются, __next__ бросает StopIteration. У асинхронного протокола те же роли, только с буквой a и с возможностью ждать: __aiter__ возвращает итератор, __anext__ — корутина, которая выдаёт следующий элемент, а конец потока объявляется исключением StopAsyncIteration.

Вся разница — в одном слове: __anext__ может делать await. Значит, пока очередное событие не пришло, корутина честно ждёт, а событийный цикл занимается остальными задачами. Напишем минимальный итератор: тики, по одному в секунду.

import asyncio

class Ticker:
    """Выдаёт числа с паузой - модель потока событий."""

    def __init__(self, n, delay):
        self.n = n          # сколько тиков выдать
        self.delay = delay  # пауза между тиками, секунд
        self.i = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.i >= self.n:
            raise StopAsyncIteration          # сигнал: поток закончился
        self.i += 1
        await asyncio.sleep(self.delay)       # "ждём данные"
        return self.i

async def main():
    async for tick in Ticker(3, 1):
        print("тик", tick)

asyncio.run(main())

тик 1
тик 2
тик 3

Разбираем механику. Строка async for tick in Ticker(3, 1) вызывает __aiter__ — здесь он возвращает сам объект; дальше на каждом витке async for дожидается __anext__ и получает число. Как только __anext__ бросил StopAsyncIteration, цикл молча поймал его и завершился — для async for это не ошибка, а штатный сигнал «данных больше нет». Пауза на каждом шаге здесь имитация: в настоящем потоке вместо sleep стояло бы ожидание сети или очереди.

Где это нужно на практике: бот читает поток апдейтов, парсер — ленту ссылок, дашборд — телеметрию приборов. Общее у всех трёх одно: элементы появляются в непредсказуемые моменты, а реагировать надо немедленно после прихода очередного. Итератор прячет всё ожидание внутри __anext__, и consumer-код остаётся наивно простым циклом — вся сложность спрятана в источнике.

Никто не запрещает ходить по итератору руками: возьми его через __aiter__ и вызывай __anext__ по одному, ловя StopAsyncIteration сам. Так делают, когда шаг цикла нужен не сразу, а по требованию — например, вынуть из потока только первые два события.

import asyncio

class Ticker:
    def __init__(self, n, delay):
        self.left = n
        self.delay = delay

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.left <= 0:
            raise StopAsyncIteration
        self.left -= 1
        await asyncio.sleep(self.delay)
        return self.left + 1

async def main():
    it = Ticker(2, 1).__aiter__()     # итератор можно взять руками
    print("шаг вручную:", await it.__anext__())
    print("шаг вручную:", await it.__anext__())
    try:
        await it.__anext__()          # третий раз - уже пусто
    except StopAsyncIteration:
        print("StopAsyncIteration: данные кончились")

asyncio.run(main())

шаг вручную: 2
шаг вручную: 1
StopAsyncIteration: данные кончились

Деталь, мимо которой проходят: __aiter__ не обязан возвращать self. По протоколу он может выдавать свежий объект-итератор — тогда контейнер остаётся многоразовым, и каждый async for начинает читать с начала. Наши классы возвращают self: объект расходуется за один проход, и второй цикл по нему получит StopAsyncIteration сразу же. Для потоков событий это честно — события не переигрываются.

И про родню исключений: StopAsyncIteration — наследник обычного StopIteration, но подменять их нельзя. Синхронный __next__ объявляет конец через StopIteration, асинхронный __anext__ — через StopAsyncIteration, и перепутать их опасно: StopIteration, вылетевший из корутины, превращается в RuntimeError. У async-протокола собственное исключение конца — не украшение, а защита от этой ловушки.

Асинхронные генераторы: поток в пять строк

Писать класс с двумя dunder-методами ради каждого потока утомительно — и Python уже придумал сокращение. Если внутри async def встречается yield, функция становится асинхронным генератором: вызов создаёт готовый async-итератор, __aiter__ и __anext__ у него уже есть, и async for ходит по нему без всяких церемоний. Это тот же приём, что и обычные генераторы с yield, только шаги умеют ждать.

import asyncio

async def telemetry(ticks):
    """Асинхронный генератор: yield внутри async def."""
    for i in range(1, ticks + 1):
        await asyncio.sleep(1)                # имитация поступления данных
        yield f"показание {i}"

async def main():
    async for msg in telemetry(3):
        print(msg, "- обработано")

asyncio.run(main())

показание 1 - обработано
показание 2 - обработано
показание 3 - обработано

Три строки тела — и полноценный поток. В реальном проекте вместо sleep здесь был бы await websocket.recv() или чтение очереди, а consumers кода бы не изменился: async for одинаково ходит и по классу с протоколом, и по генератору. Поэтому правило простое: один поток на один раз — генератор; поток, который хранит состояние и умеет останавливаться по-разному, — класс.

Проверим это честно: замерим, сколько длится цикл, в котором и поток, и «обработка» паузят по секунде. Задержки складываются, потому что шаги идут по очереди — время печатаем целыми секундами через round.

import asyncio

async def telemetry(ticks, interval):
    for i in range(1, ticks + 1):
        await asyncio.sleep(interval)
        yield i

async def main():
    loop = asyncio.get_running_loop()
    start = loop.time()
    async for i in telemetry(2, 1):
        await asyncio.sleep(1)                # "тяжёлая" обработка
        print("элемент", i, "обработан")
    print("прошло секунд:", round(loop.time() - start))

asyncio.run(main())

элемент 1 обработан
элемент 2 обработан
прошло секунд: 4

Поток можно бросить: break и бесконечные источники

Настоящие потоки событий бесконечны: телеметрия течёт, пока жив бот. Асинхронному генератору ничего не стоит выдавать элементы вечно — вопрос лишь в том, кто остановит чтение. Ответ — break: он выходит из async for и закрывает генератор. И главное свойство такого дизайна: элемент вычисляется только когда запрошен. До break тело третьего витка не дошло — значит, четвёртое событие никто и не ждал.

import asyncio

async def sensor():
    """Бесконечный поток: выдаёт числа, пока его читают."""
    i = 0
    while True:
        await asyncio.sleep(1)
        i += 1
        yield i

async def main():
    async for value in sensor():
        print("взяли", value)
        if value >= 3:
            break                  # единственный выход из бесконечного потока

asyncio.run(main())

взяли 1
взяли 2
взяли 3

Кстати, break не просто прекращает цикл: при выходе из async for генератор закрывается, и внутри него поднимается особый GeneratorExit. Потребителю это незаметно, а автору генератора даёт точку уборки — открытые в потоке ресурсы освобождают в finally, и они закроются как при break, так и при полном вычитывании потока.

Поток событий по тикам: собираем всё вместе

Теперь соберём итератор, на который похожи настоящие ленты событий: события приходят вразнобой — что-то сразу, что-то с задержкой — а consumer получает их строго по мере поступления, один за другим. Класс хранит список событий с паузами и отдаёт их по одному; между событиями — честное ожидание.

import asyncio

class EventStream:
    """События приходят вразнобой - итератор отдаёт их по одному."""

    def __init__(self, events):
        self.events = list(events)   # пары (пауза, событие)
        self.pos = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.pos >= len(self.events):
            raise StopAsyncIteration
        delay, value = self.events[self.pos]
        self.pos += 1
        await asyncio.sleep(delay)   # имитация ожидания события
        return value

async def main():
    stream = EventStream([
        (1, "пинг"),
        (2, "заказ"),
        (1, "логин"),
    ])
    async for event in stream:
        print("событие:", event)

asyncio.run(main())

событие: пинг
событие: заказ
событие: логин

Обрати внимание: consumer не знает, когда придёт следующее событие, и не обязан знать. Он просто читает поток — вся работа по ожиданию спрятана в __anext__. Именно так устроен async for по сообщениям бота в aiogram или по строкам ответа сервера: подписался, ходишь циклом, обрабатываешь по одному. А если события надо не читать по одному, а собирать пачкой и сравнивать, кто быстрее, — это территория gather и as_completed из урока 8.

Когда поток — не единственный источник данных, его стыкуют с очередью: продюсер кладёт события в asyncio.Queue из урока 13, а генератор читает её через await queue.get(). Получается тот же async for, но с буфером внутри — события не теряются, даже если обработка подтормаживает. Ну а следующий шаг по протоколам — асинхронные контекст-менеджеры из урока 16: потоки почти всегда открывают ресурс, который надо корректно закрыть.

Стандартная библиотека и экосистема дают готовые async-итераторы на все случаи: чтение вывода подпроцесса, асинхронные файлы в aiofiles, страницы результатов API, отдающие данные постранично. Всякий раз, когда в документации видишь пометку async iterator или метод __anext__, за ней стоит ровно тот протокол, который ты только что собрал руками.

# Фрагмент без запуска: сеть в песочнице недоступна.
# С вебсокетом async for выглядит ровно так же, как в наших примерах:
async for message in websocket:
    await handle(message)      # message приходит, когда пришёл - не чаще

# И с aiohttp при чтении ответа кусками:
async for chunk in resp.content:
    save(chunk)

Что дальше

Ты умеешь читать и писать потоки: протокол __aiter__/__anext__, StopAsyncIteration как конец, асинхронные генераторы как короткий путь, break как тормоз. Осталось научиться гарантированно убирать за ресурсами, которые эти потоки открывают, — этим занимается async with, герои следующего урока. А сам async for ты встретишь в aiogram и вебсокетах: хендлеры бота — тот же цикл по потоку апдейтов, только спрятанный за декораторами.

Асинхронная итерация — это обычный for, которому разрешили ждать: элемент отдаётся тогда, когда он появился, а не когда захотелось списку.