Python : 비동기 이터레이터, 비동기 제네레이터

들어가기 전

이 글은 파이썬 비동기 라이브러리 asyncio를 공부하며 작성한 글입니다.

 

---

3.7.0 이터레이터

비동기 이터레이터를 시작하기 전에 이터레이터가 무엇인지를 간략히 살펴보고자 한다. 파이썬에서는 다음이 구현된 녀석을 이터레이터라고 한다.

• __iter__() 매직 메서드가 구현됨. 
• __next__() 매직 메서드가 구현됨 

위 두 개의 메서드가 구현되었다면, 그 클래스는 이터레이터가 된다. 그리고 이터레이터는 for 문 같은 곳에서 자유롭게 iterate 할 수 있게 된다. 아래 코드로 예를 들 수 있다. 

class A:
    def __init__(self):
        self.x = -1

    def __iter__(self):
        self.x = 0
        return self

    def __next__(self):
        if self.x > 2:
            raise StopIteration
        else:
            self.x += 1
            return self.x

for i in A():
    print(i)
    
    
>>>>
1
2

각 코드의 내용을 살펴보면 다음과 같다.

__iter__()

이 메서드는 반드시 iterable 객체를 반환해야한다. iterable 객체는 __next__() 메서드가 구현된 객체다. 위 코드에서는 __next__()가 구현된 자기 자신을 iterable 객체로 반환한다.

__next__()

Iterable 객체는 반복문을 통해서 StopIteration이 발생될 때까지 계속 호출될 것이다. __next__() 메서드를 구현할 때는 다음 두 가지를 고려하면 된다.

• 반복 할 때 마다 반환할 값을 생성해서 반환한다. 
• 반복이 끝날 때 StopIteration을 발생시켜야 한다.

 

실행 결과 1,2가 출력되는 것을 살펴볼 수 있다. 위의 함수는 동기 형태로 구현된 Iterator다. 그렇다면 비동기 Iterator는 어떻게 구현해 볼 수 있고, 어떻게 사용할 수 있을까?

---

3.7.1 비동기 이터레이터 async for

이터레이터와 마찬가지로 비동기 이터레이터도 사용하기 위해 필요한 코드 정의가 있다.

• def __aiter__()를 구현해야한다. (이 때 async def가 아니다)
• __aiter__()는 async def __anext__()를 구현한 객체를 반환해야 한다. 
• __anext__()는 반복의 각 단계에 대한 값을 반환하고, 반환이 끝나면 StopAsyncIteration을 발생시켜야 한다. 

어떻게 동작하는지 살펴보기 위해서 아래에 간단한 예제를 작성해서 알아보자. Redis에서 여러 개의 키에 대해 반복적으로 수행하며 값을 확인할 때, 요청 시점에 데이터를 가져오는 경우를 가정한다. 

import asyncio
import aioredis

async def create_redis(h): pass
async def do_something_with(value): pass

async def main():
    redis = await create_redis(('localhost', 6379))
    keys = ['Americas', 'Africa', 'Europe', 'Asia']

    async for value in OneAtATime(keys):
        await do_something_with(value)

class OneAtATime:
    def __init__(self, redis, keys):
        self.redis = redis
        self.keys = keys
    def __aiter__(self):
        self.ikeys = iter(self.keys)
        return self
    async def __anext__(self):
        try:
            key = next(self.ikeys)
        except StopIteration: ## --> 키 반복이 완료되면 StopIteration 발생함.
            raise StopAsyncIteration

        value = await self.redis.get(key) ## --> 비동기로 데이터 가져옴
        return value

asyncio.run(main())

코드는 위와 같다. 아래에서 코드를 하나씩 더 살펴보면 다음과 같다. 

async for

• 비동기 이터레이터를 이용해서 데이터를 받아오고 싶다면 async for을 이용해야한다. async 키워드를 했기 때문에 반복 중에 다음 데이터를 얻기 전까지 반복 자체를 일시정지할 수 있게 된다.
• 다음 데이터를 얻을 때까지의 일시정지는 __anext__() 메서드 내부에 선언된 await 키워드를 통해 구현된다. 

__aiter__()

• __aiter__() 메서드에서 반복을 수행하기 위한 준비 작업을 한다. Key 목록(keys)으로 일반적인 이터레이터인 ikeys를 생성한다.
• OneAtATime 클래스는 async def __anext__()를 구현한 iterable한 객체다. 따라서 자기 자신을 반환한다. 

__anext__()

• self.ikeys는 iter(keys)를 통해서 만들어진 변수다. 즉, 이터레이터다. 이터레이터가 반복이 완료되면 StopIteration이 발생한다. 
• __anext__() 메서드에서는 StopIteration이 발생하면 그것을 Catch해서 StopAsyncIteration 에러를 던져주도록 한다. 이렇게 하면 비동기 이터레이터의 실행이 종료된다. 

 

위 코드에서는 async def __anext__() 내부에서 또 다른 코루틴을 호출하여 코루틴 체이닝이 되도록 작성되었다. 또한 이렇게 구성된 비동기 이터레이터는 for 루프의 편의성은 유지하면서, 데이터를 가져오는 I/O를 반복수행하는 동작을 비동기로 처리할 수 있게 되었다. 즉, async for를 이용해서 for를 비동기로 처리할 수 있게 되었다. 비동기로 처리할 수 있다는 말은 특정한 데이터를 가져올 때 까지 이 함수의 실행을 대기하고 다른 함수를 실행할 수 있다는 것을 의미한다. 

그런데 한 가지 단점이 있다. 구현해야 할 코드의 양이 너무 많다는 점이다. 파이썬은 이 부분을 개선할 수 있도록 비동기 제네레이터를 제공하며 이 부분을 더욱 간단하게 구현할 수 있게 해준다.

---

3.8 비동기 제네레이터 사용. 

비동기 제네레이터를 사용하면 비동기 이터레이터를 사용하는 것에 비해 더욱 손쉽게 코드로 처리할 수 있다. 비동기 제네레이터에서 대해서 간단히 하나씩 알아보면 다음과 같다. 

• 비동기 제네레이터는 async def로 선언된 함수이며, 내부에 yield 키워드를 가지고 있다.
• 코루틴(비동기 함수)과 제네레이터(yield 키워드)는 완전히 다른 개념이다. 
• 비동기 제네레이터는 일반 제네레이터와 매우 유사하게 동작한다. 
• 반복 수행 시, 일반 제네레이터는 for를 이용해서 반복한다. 비동기 제네레이터는 async for를 이용해서 반복한다. 

비동기 제네레이터는 내부적으로 yield 키워드를 사용해서 Context를 기억하고 반환하며, 내부에서 코루틴 체이닝이 가능하다. 따라서 비동기 제네레이터를 선언할 때는 async def로 선언하되 내부에 반드시 yield 키워드를 가져야 한다. 비동기 제네레이터를 for 루프를 이용해서 사용하고 싶다면 async for를 이용하면 된다! 

async def create_redis(h): pass
async def do_something_with(value): pass

async def main():
    redis = await create_redis(('localhost', 6379))
    keys = ['Americas', 'Africa', 'Europe', 'Asia']

    async for value in one_at_a_time(redis, keys):
        await do_something_with(value)

async def one_at_a_time(redis, keys):
    for k in keys:
        yield await redis.get(k)

코드 차이

비동기 이터레이터에서는 클래스를 하나 만들고 __aiter()__ / __anext()__ 메서드를 하나 구현해야했다. 이것은 코드가 길어질 수 있는 요소인데, 비동기 제네레이터를 이용하면 위와 같이 간략하게 만들 수 있다. 

async for 루프

• 비동기 제네레이터를 사용할 때는 async for를 이용해서 for 루프를 돌려야 한다. async for 루프는 제네레이터가 끝날 때까지 반복한다. 
• 위 코드에서 비동기 제네레이터는 key라는 변수를 한번씩 다 탐색하는 동안만 사용된다. 그리고 내부적으로 redis.get()이라는 메서드를 await 한다.
• 정리하면 async for를 이용해서 비동기 제네레이터를 돌리면, await redis.get()이 될 때까지 대기한 후에 필요한 로직을 수행한다는 것이다.

아래에서 실제로 간단히 사용해 볼 수 있는 예시를 하나 개인적으로 생성했다. 

import asyncio
import aiohttp

async def main():

    urls = ['http://google.com', 'http://naver.com', 'http://daum.net', 'http://github.com']
    async with aiohttp.ClientSession() as session:
        async for response in one_at_a_time(session, urls):
            print(f'{response=}')
    print('end job')

async def one_at_a_time(session, urls):
    for url in urls:
        yield await session.request('get', url)


asyncio.run(main())

코드

• 비동기 Http 요청을 처리해주는 aiohttp 코드를 작성했다. 
• async with를 이용해서 네트워크 인터페이스를 사용하고 종료하도록 했다. 
• 각 url에 대해서 요청을 보내고 비동기로 데이터를 받아오고, 그것을 response에 저장한다.

 

정리하면 다음과 같다.

• 비동기 이터레이터를 구현하기 위해서는 __aiter()__ / __anext()__를 구현해야 하기 때문에 코드의 양이 많아진다.
• 비동기 이터레이터를 구현해야한다면, 비동기 제네레이터를 이용해서 좀 더 코드를 간결하게 하는 방법이 있기 때문에 이것을 사용하는 방법도 고려해야한다. 

 

---

3.9 비동기 컴프리헨션

async for를 이용해서 for 루프를 돌릴 수 있다면, 당연한 말이지만 비동기 컴프리헨션(Comprehension) 역시 가능하다. 예를 들어 위의 코드를 아래와 같이 바꿀 수 있다.

async def main():
    urls = ['http://google.com', 'http://naver.com', 'http://daum.net', 'http://github.com']
    async with aiohttp.ClientSession() as session:
        responses = [response async for response in one_at_a_time(session, urls)]
        print(f'{responses=}')
    print('end job')

async def one_at_a_time(session, urls):
    for url in urls:
        yield await session.request('get', url)

asyncio.run(main())

이 밖에도 한 가지 알아두면 좋을 부분은 다음과 같다.

• 컴프리헨션 내에서 await 키워드를 사용할 수 있다.

위는 당연한 이야기다. 비동기 컴프리헨션이 가능하게 하는 것은 async for를 통해서 for 루프를 비동기로 돌릴 수 있기 때문이다. async for를 통해서 공급되는 값이 코루틴이라면 당연한 이야기지만, await를 통해서 값을 얻을 수 있다. 예를 들면 아래와 같은 코드도 가능하다.

async def f1(x):
    await asyncio.sleep(1)
    return x + 100

async def factory(n):
    for x in range(n):
        yield f1(x) ## --> 리턴되는 것은 코루틴이다.


async def main():
    result = [await f async for f in factory(5)]
    print(f'{result=}')

asyncio.run(main())
>>>
result=[100, 101, 102, 103, 104]

 

 

 

---

정리

• 비동기 이터레이터, 제네레이터, 컴프리헨션은 async for를 이용해서 비동기 반복 객체를 처리할 수 있다.
• 비동기 이터레이터는 __aiter()__ / async def __anext()__ 를 구현해야 하기 때문에 코드의 구현량이 많을 수 있다.
• 비동기 제네레이터는 async def로 선언되고 내부에 yield 키워드를 가지는 함수다. 
• 비동기 컴프리헨션은 내부적으로 await 키워드를 가질 수 있다. await 키워드는 전달되는 인자가 코루틴일 때만 가능하다.