Async generators и async iterators в Python

Async generators — способ лениво производить данные в async контексте. Идеальны для стриминга, пагинации и обработки больших датасетов.

Async generator — yield внутри async def

import asyncio, httpx

async def paginate(client, base_url, pages=5):
    for page in range(1, pages + 1):
        r = await client.get(f"{base_url}?page={page}")
        data = r.json()
        for item in data['results']:
            yield item          # yield в async def = async generator

async def main():
    async with httpx.AsyncClient() as client:
        async for item in paginate(client, "https://api.example.com/posts"):
            print(item['title'])    # обрабатываем один за раз, не всё сразу

Без async generators пришлось бы загружать все страницы в память. Здесь каждый элемент обрабатывается сразу после получения.

Стриминг LLM ответа через async generator

from anthropic import AsyncAnthropic

async def stream_response(prompt: str):
    client = AsyncAnthropic()
    async with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        async for text in stream.text_stream:
            yield text              # каждый токен — отдельный yield

async def main():
    async for chunk in stream_response("Напиши стихотворение"):
        print(chunk, end="", flush=True)

Типизация async генераторов

from typing import AsyncGenerator, AsyncIterator

# AsyncGenerator[YieldType, SendType]
async def number_stream(n: int) -> AsyncGenerator[int, None]:
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

# AsyncIterator — если не нужен SendType
async def events() -> AsyncIterator[dict]:
    while True:
        event = await fetch_event()
        yield event

aiter и anext — кастомный итератор

class DatabaseCursor:
    def __init__(self, query: str):
        self.query = query
        self._rows = []
        self._index = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self._index >= len(self._rows):
            # Загружаем следующую порцию из БД
            self._rows = await fetch_batch(self.query, self._index)
            if not self._rows:
                raise StopAsyncIteration
        row = self._rows[self._index]
        self._index += 1
        return row