[Python] anext() 함수 사용법

anext() 에 대하여 자세하게 알아봅시다

anext()

anext()는 Python 3.10부터 추가된 비동기 이터레이터용 내장 함수입니다. 동기 이터레이터에서 next()를 쓰듯, 비동기 이터레이터(async iterator)로부터 다음 값을 await 키워드와 함께 받을 때 사용합니다. (PEP 525 참고)

• 형식:

  • await anext(async_iterable)
  • await anext(async_iterable, default) (StopAsyncIteration 발생 시 기본값 반환)

• 사용 목적:

  • 비동기 제너레이터나 비동기 이터레이터로부터 값을 하나씩 가져오고 싶을 때 사용합니다.

기본 사용법 예시

1. 기본 사용법: 비동기 이터레이터에서 값 추출

import asyncio

async def my_gen():
    for i in range(3):
        yield i

async def main():
    agen = my_gen()
    print(await anext(agen))  # 0
    print(await anext(agen))  # 1

asyncio.run(main())

2. 기본값 사용

import asyncio

async def my_gen():
    yield 1

async def main():
    agen = my_gen()
    print(await anext(agen, '끝!'))  # 1
    print(await anext(agen, '끝!'))  # '끝!'

asyncio.run(main())

3. StopAsyncIteration 예외 처리 없이 사용 (default 값 없을 때)

import asyncio

async def empty_gen():
    if False:
        yield

async def main():
    agen = empty_gen()
    try:
        print(await anext(agen))
    except StopAsyncIteration:
        print('No more items!')

asyncio.run(main())

고급 사용법 예시

1. 여러 이터레이터를 순차적으로 처리

import asyncio

async def ag1():
    for i in range(2):
        yield f'ag1-{i}'
async def ag2():
    for i in range(2):
        yield f'ag2-{i}'

async def main():
    gens = [ag1(), ag2()]
    for g in gens:
        while True:
            item = await anext(g, None)
            if item is None:
                break
            print(item)

asyncio.run(main())

2. 비동기적으로 여러 비동기 이터레이터 병렬 처리

import asyncio

async def agen(n):
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

async def fetch_all(*gens):
    tasks = [anext(g, 'DONE') for g in gens]
    return await asyncio.gather(*tasks)

async def main():
    g1, g2 = agen(2), agen(3)
    print(await fetch_all(g1, g2))  # [0, 0]
    print(await fetch_all(g1, g2))  # [1, 1]
    print(await fetch_all(g1, g2))  # ['DONE', 2]

asyncio.run(main())

3. 커스텀 비동기 이터레이터에서 anext() 사용

import asyncio

class Counter:
    def __init__(self, stop):
        self.cur = 0
        self.stop = stop
    def __aiter__(self):
        return self
    async def __anext__(self):
        if self.cur < self.stop:
            await asyncio.sleep(0.05)
            result = self.cur
            self.cur += 1
            return result
        else:
            raise StopAsyncIteration

async def main():
    counter = Counter(2)
    print(await anext(counter))  # 0
    print(await anext(counter))  # 1
    print(await anext(counter, 'END'))  # 'END'

asyncio.run(main())

총평

anext()는 비동기 프로그래밍에서 비동기 이터레이터(예: async generator)로부터 값을 순차적으로 받아올 때 매우 유용하게 쓰입니다. 기존의 next() 함수와 거의 동일하게 동작하지만, awaitable 객체를 다룬다는 차이가 있습니다. 기본값을 제공해서 StopAsyncIteration 예외를 처리하는 것도 편리한 점입니다. 비동기 반복문(async for)에서 벗어나 값을 하나하나 뽑거나, 여러 이터레이터를 동적으로 조작할 때 특히 활용도가 높습니다.