Async 반복 — async for / __aiter__ / __anext__

반복 프로토콜의 async 사촌

일반 반복 — __iter__ + __next__. async 반복 — __aiter__ + __anext__. 차이점은 __anext__ 가 coroutine — 다음 값 만들기 전에 뭔가 await 가능. consumer 쪽은 async for x in stream:. 이 프로토콜이 streaming 응답, async DB 커서, websocket 메시지 루프 — 각 값 만드는 데 I/O 가 필요한 모든 거 동력.

Async generator — 쉬운 길

일반 generator 가 클래스 기반 iterator 를 대부분 대체하듯, async generator 가 클래스 기반 async iterator 를 대부분 대체. async def 함수에 yield 사용. 함수가 async generator 가 됨. async for 가 소비. 각 yield 전에 await 가능. cwkPippa 의 어댑터들이 이 패턴 — 청크가 네트워크로 들어와, 처리되고, yield 로 stream out.

Async 컴프리헨션

[x async for x in stream] — async list 컴프리헨션. (x async for x in stream) — async generator expression. 둘 다 *오직* async def 안에서만. 이미 async 컨텍스트면 편리.

원칙: sync 반복과 async 반복은 다른 프로토콜. 일반 iterable 에 async for X, async iterable 에 for X. 에러 메시지 명확 ("not async iterable" / "not iterable") — 프로토콜 알면 디코딩 시간 절약.

실전 코드에 어디 나오나

웹 프레임워크의 응답 청크 streaming. 데이터베이스 커서가 row 도착 즉시 yield. AI SDK (Claude Agent SDK 같은) 가 토큰, 도구 호출, 라이프사이클 이벤트 yield. WebSocket 메시지 루프. "나타나는 즉시 각 항목 처리" 라고 말할 만한 모든 곳. concurrency 트랙이 asyncio 깊이 다룸 — 이 lesson 은 반복 모양만.

Code

Async generator — 단순한 모양·python

import asyncio

async def ticker(n):
    for i in range(n):
        await asyncio.sleep(0.1)        # I/O 좀
        yield i

async def main():
    async for x in ticker(3):
        print(x)
# 0
# 1
# 2  (각 ~100ms 간격)

asyncio.run(main())

async for — consumer 쪽·python

import asyncio

async def chunks():
    for piece in ["hello ", "world ", "from ", "pippa"]:
        await asyncio.sleep(0.05)
        yield piece

async def collect():
    full = ""
    async for chunk in chunks():
        full += chunk
    return full

result = asyncio.run(collect())
print(result)                   # 'hello world from pippa'

Async 컴프리헨션·python

import asyncio

async def naturals(limit):
    for n in range(1, limit + 1):
        await asyncio.sleep(0.01)
        yield n

async def main():
    # async list 컴프리헨션
    squares = [x*x async for x in naturals(5)]
    print(squares)              # [1, 4, 9, 16, 25]

    # async generator expression
    total = sum([x*x async for x in naturals(5)])
    print(total)                # 55

asyncio.run(main())

클래스 기반 async iterator (state 필요할 때)·python

import asyncio

class DelayedRange:
    def __init__(self, n, delay=0.05):
        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
        await asyncio.sleep(self.delay)
        v = self.i
        self.i += 1
        return v

async def main():
    async for x in DelayedRange(4):
        print(x)

asyncio.run(main())

잘못된 프로토콜 — 명확한 에러·python

import asyncio

async def main():
    # 일반 iterable 에 async for — 에러
    try:
        async for x in [1, 2, 3]:
            pass
    except TypeError as e:
        print(e)                # 'list' object is not async iterable

    # async iterable 에 for — 에러
    async def stream():
        yield 1

    try:
        for x in stream():
            pass
    except TypeError as e:
        print(e)                # 'async_generator' object is not iterable

asyncio.run(main())

Exercise

Async generator tick(every, count) 작성 — 정수 0, 1, ..., count-1 yield, 각 사이 every 초의 await asyncio.sleep. 그 다음 async for 로 소비하는 async 함수, 각 값과 경과 시간 출력. 그 다음 async list 컴프리헨션 으로 list 에 모으고 마지막에 다 출력하는 버전. asyncio.run 으로 실행.

Async 반복 — async for / aiter / anext

반복 프로토콜의 async 사촌

Async generator — 쉬운 길

Async 컴프리헨션

실전 코드에 어디 나오나

Code

External links

Exercise

Progress

댓글 0