Async Generators and Async Iterators in Python

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 in 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'])    # process one at a time, not everything at once

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              # each token is a separate yield

async def main():
    async for chunk in stream_response("Write a poem"):
        print(chunk, end="", flush=True)

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 — when SendType isn't needed
async def events() -> AsyncIterator[dict]:
    while True:
        event = await fetch_event()
        yield event

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):
            # Load next batch from DB
            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