1. 제너레이터 만들기
제너레이터는 파이썬에서 고성능이면서도 메모리를 적게 사용하는 반복을 위한 방법이다.
제너레이터는 한 번에 하나씩 구성요소를 반환해주는 이터러블을 생성해주는 객체이다. 제너레이터를 사용하는 주요 목적은 메모리를 절약하는 것이다. 거대한 요소를 한꺼번에 메모리에 저장하는 대신 특정 요소를 어떻게 만드는지 아는 객체를 만들어서 필요할 때마다 하나씩 가져오는 것이다.
이 기능은 lazy computation 을 통해 가능하다.
1-1. 제너레이터 개요
대규모의 구매 정보에서 최저 판매가, 최고 판매가, 평균 판매가를 구하는 예제가 있다.
문제를 단순화하여 단 두 개의 필드만을 가진 csv 가 있다고 가정한다.
<purchase_date>, <price>
모든 구매 정보를 받아 필요한 지표를 구해주는 객체를 만들어본다.
class PurchasesStats:
def __init__(self, purchases):
self.purchases = iter(purchases)
self.min_price: float = None
self.max_price: float = None
self._total_purchases_price: float = 0.0
self._total_purchases = 0
self._initialize()
def _initialize(self):
try:
first_value = next(self.purchases)
except StopIteration:
raise ValueError("no values provided")
self.min_price = self.max_price = first_value
self._update_avg(first_value)
def process(self):
for purchase_value in self.purchases:
self._update_min(purchase_value)
self._update_max(purchase_value)
self._update_avg(purchase_value)
return self
def _update_min(self, new_value: float):
if new_value < self.min_price:
self.min_price = new_value
def _update_max(self, new_value: float):
if new_value > self.max_price:
self.max_price = new_value
@property
def avg_price(self):
return self._total_purchases_price / self._total_purchases
def _update_avg(self, new_value: float):
self._total_purchases_price += new_value
self._total_purchases += 1
def __str__(self):
return (
f"{self.__class__.__name__}({self.min_price}, "
f"{self.max_price}, {self.avg_price})"
)
위 객체는 모든 구매 정보를 받아서 필요한 계산을 한다. 아래와 같이 모든 정보를 로드해서 어딘가에 담아 반환해주는 함수를 만들 수 있다.
def _load_purchases(filename):
purchases = []
with open(filename) as f:
for line in f:
*_, price_raw = line.partition(",")
purchases.append(float(price_raw))
return purchases
위 코드는 정상적인 결과를 반환한다. 파일에서 모든 정보를 읽어서 리스트에 저장하는 방식이다. 하지만 파일이 상당히 많은 데이터를 가지고 있다면 로드하는데 시간이 오래 걸리고 메인 메모리에 담지 못할 만큼 큰 데이터일 수도 있다.
그러나 코드에서 보면 한 번에 하나의 데이터 만을 사용하고 있다는 걸 알 수 있다. 따라서 굳이 모든 데이터를 읽어와 메모리에 담고 있을 필요가 없다.
이에 대한 해결책은 제너레이터를 만드는 것이다. 파일의 전체 내용을 리스트에 저장하는 대신에 필요한 값만 그때그때 가져오는 것이다.
def load_purchases(filename):
with open(filename) as f:
for line in f:
*_, price_raw = line.partition(",")
yield float(price_raw)
이렇게 수정하면 메모리 사용량이 크게 떨어진다. 결과를 담을 리스트가 필요 없어졌으며 return 문 또한 사라졌다.
이 경우 load_pruchases 함수를 제너레이터 함수 또는 단순히 제너레이터 라고 한다.
파이썬에서 어떤 함수라도 yield 키워드를 사용하면 제너레이터 함수가 된다. yield 가 포함된 함수를 호출하면 제너레이터 인스턴스를 만든다.
>>> load_purchases("file")
(generator object load_purchases at ..>
모든 제너레이터 객체는 이터러블이다.
1-2. 제너레이터 표현식
제너레이터를 사용하면 많은 메모리를 절약할 수 있으며 이터레이터이므로 리스트나 튜플, 세트처럼 많은 메모리를 필요로 하는 이터러블이나 컨테어너의 대안이 될 수 있다.
또한 컴프리헨션에 의해 정의될 수 있는 리스트나 세트, 사전처럼 표현식으로 정의할 수 있다. 제너레이터 표현식은 sum() 이나 max() 와 같이 이터러블 연산이 가능한 함수에 직접 전달할 수도 있다.
>>> sum(x**2 for x in range(10))
285
2. 이상적인 반복
2-1. 관용적인 반복 코드
내장함수인 enumerate() 는 이터러블을 입력 받아서 인덱스 번호와 원본의 원소를 튜플 형태로 변환하여 enumerate 객체를 반환한다.
>>> list(enumerate("abcdef"))
[(0, 'a'), (1, 'b'), (2, 'c'), (3, 'd'), (4, 'e'), (5, 'f')]
좀 더 저수준에서 이와 유사한 객체를 만들어본다. 만들어볼 객체는 단순히 시작 값을 입력하면 무한 시퀀스를 만드는 역할을 한다.
class NumberSequence:
def __init__(self, start=0):
self.current = start
def next(self):
current = self.current
self.current += 1
return current
이 인터페이스에 기반을 두어 클라이언트를 작성하면 명시적으로 next() 함수를 호출해야 한다.
>>> seq = NumberSequence()
>>> seq.next()
0
>>> seq.next()
1
>>> seq2 = NumberSequence(10)
>>> seq2.next()
10
>>> seq2.next()
11
그러나 이 코드로 enumerate() 함수를 사용하도록 재작성할 수는 없다. 왜냐하면 파이썬의 for 루프를 사용하기 위한 인터페이스를 지원하지 않기 때문이다. 이는 다시 말해서 이터러블 형태의 파라미터로는 사용할 수 없다는 것을 뜻한다.
>>> list(zip(NumberSequence(), "abcdef"))
Traceback (most recent call last):
File "<input>", line 1, in <module>
TypeError: zip argument #1 must support iteration
NumberSequence 가 반복을 지원하지 않는다. 이를 해결하려면 __iter__() 매직 메서드를 구현하여 객체가 반복 가능하게 만들어야 한다. 또한 next() 메서드를 수정하여 __next__ 매직 메서드를 구현하면 객체는 이터레이터가 된다.
class SequenceOfNumbers:
def __init__(self, start=0):
self.current = start
def __next__(self):
current = self.current
self.current += 1
return current
def __iter__(self):
return self
이렇게 하면 요소를 반복할 수 있을 뿐만 아니라 .next() 메서드를 호출할 필요도 없다. 왜냐하면 __next__() 메서드를 구현했으므로 next() 내장 함수를 사용할 수 있기 때문이다.
>>> list(zip(SequenceOfNumbers(), "abcdef"))
[(0, 'a'), (1, 'b'), (2, 'c'), (3, 'd'), (4, 'e'), (5, 'f')]
>>> seq = SequenceOfNumbers(100)
>>> next(seq)
100
>>> next(seq)
101
# next() 함수
next() 내장 함수는 이터러블을 다음 요소로 이동시키고 기존의 값을 반환하며 이터레이터가 더 이상의 값을 가지고 있지 않다면 StopIteration 예외가 발생한다.
>>> word = iter("Hello")
>>> next(word)
'H'
>>> next(word)
'e'
...
>>> next(word)
Traceback (most recent call last):
File "<input>", line 1, in <module>
StopIteration
예외는 반복이 끝났음을 의미하며 사용할 수 있는 요소가 더 이상 없음을 나타낸다.
이 문제를 해결하고 싶다면 StopIteration 예외를 캐치하는 것 외에도 next() 함수의 두 번째 파라미터에 기본 값을 제공할 수도 있다. 이 값을 제공하면 StopIteration 을 발생시키는 대신 기본 값을 반환한다.
>>> next(word, "default value")
'default value'
# 제너레이터 사용하기
앞의 코드는 제너레이터를 사용하여 훨씬 간단하게 작성할 수 있다. 제너레이터를 사용하면 클래스를 만드는 대신 다음과 같이 필요한 값을 yield 하는 함수를 만들면 된다.
def sequence(start=0)
while True:
yield start
start += 1
함수 본문의 yield 키워드가 해당 함수를 제너레이터로 만들어 준다. 이 함수는 제너레이터이기 때문에 무한 루프를 사용해도 완벽하게 안전하다. 제너레이터 함수가 호출되면 yield 문장을 만나기 전까지 실행된다. 그리고 값을 생성하고 그 자리에서 멈춘다.
>>> seq = sequence(10)
>>> next(seq)
10
>>> next(seq)
11
>>> list(zip(sequence(), "abcdef"))
[(0, 'a'), (1, 'b'), (2, 'c'), (3, 'd'), (4, 'e'), (5, 'f')]
# itertools
이터러블로 작업하면 코드가 파이썬 자체와 더 잘 어울리는 장점이 있다. 왜냐하면 이터레이션이 언어의 중요한 컴포넌트이기 때문이다. 또한 itertools 모듈을 사용하면 그 기능을 온전히 활용할 수 있다. 위의 sequence() 제너레이터는 itertools.count() 와 상당히 유사하다.
이터레이터, 제너레이터, itertools 와 관련하여 가장 멋진 것 중에 하나는 이들을 서로 연결하여 새로운 객체를 만들 수 있다는 것이다.
예를 들어 처음 예제를 다시 살펴보면, 만약 특정 기준을 넘은 값에 대해서만 연산을 하려면 어떻게 해야 할까? 가장 간단한 방법은 반복문 안에 조건문을 추가하는 것이다.
# ...
def process(self):
for purchase_value in self.purchases:
if purchase > 1000.0:
...
이것은 파이썬스럽지 않을 뿐 아니라 지나치게 엄격하다. 엄격하다는 것은 나쁜 코드를 의미한다. 수정사항에 유연하게 대처할 수 없음을 의미한다. 만약 기준 수치가 변경된다면 파라미터로 전달해야 할까? 만약 파라미터가 둘 이상 필요하다면? 만약 조건 기준 이하가 되면? 람다 함수를 써야 하는가?
사실 이번 객체가 이 질문에 답변할 필요가 없다. 이 객체의 고유 책임은 구매 이력에 대해 잘 정의된 지표 값을 계산하고 출력하는 것뿐이다. 이러한 요구 사항을 이번 객체에 반영하는 것 자체가 실수다.
클린 코드는 융통성이 있어야 하고 외부 요인에 결합력이 높아서는 안 된다. 위의 요구 사항은 다른 곳에서 해결되어야 한다.
코드를 수정하는 대신 그대로 유지하고 클라이언트 클래스의 요구사항이 무엇이든 그에 맞게 필터링하여 새로운 데이터를 만든다고 가정해본다. 예를 들어 1000개 넘게 구매한 이력의 처음 10개만 처리하려고 한다면 다음과 같이 할 수 있다.
>>> from itertools import islice
>>> purchases = islice(filter(lambda p: p > 1000.0, purchases), 10)
>>> stats = PurchasesStats(purchases).process()
이런 식으로 필터링을 해도 메모리의 손해는 없다. 왜냐하면 모든 것이 제너레이터이므로 지연 평가(lazy evaluation)된다. 즉, 마치 전체에서 필터링한 값으로 연산을 한 것처럼 보이지만, 실제로는 하나씩 가져와서 모든 것을 메모리에 올리 필요가 없는 것이다.
# 이터레이터를 사용한 코드 간소화
① 여러 번 반복하기
이 장의 맨 처음 살펴본 예제를 훨씬 간소화할 수 있다.
def process_purchases(purchases):
min_, max_, avg = itertools.tee(purchases, 3)
return min(min_), max(max_), median(avg)
이 예제에서 itertools.tee 는 원래의 이터러블을 세 개의 새로운 이터러블로 분할한다. 그리고 구매 이력을 세 번 반복할 필요 없이 분할된 이터러블을 사용해 필요한 연산을 수행한다.
purchases 파라미터에 다른 이터러블 객체를 넘기면 itertools.tee 함수 덕분에 원했던 것처럼 오직 한 번만 순환하는 것을 확인할 수 있다. 또한 비어있는 시퀀스를 넘기면 min() 함수에서 ValueError 를 발생시킬 것이므로 따로 ValueError 예외를 발생시키지 않아도 된다.
반복을 여러 번 해야 하는 경우 itertools.tee 를 사용한다. tee 함수는 가장 앞서 나가는 for 문을 기준으로 위치가 이동하는 셈이다. 따라서 for 문이 2개 있다고 해서 전체 데이터가 갑자기 2배로 복사되는 것은 아니지만, 적어도 어느 인덱스에서의 요소에 대해서는 2배의 메모리를 사용하게 된다. 개별 요소의 크기가 작다면 상관 없겠지만 개별 요소가 크고 이터러블을 여러 개 복사해야 한다면 유의해야 한다.
② 중첩 루프
경우에 따라 1차원 이상을 반복해서 값을 찾아야 할 수 있다. 가장 쉽게 해결하는 방법으로 중첩 루프가 있다. 값을 찾으면 순환을 멈추고 break 키워드를 호출해야 하는데 이런 경우 한 단계가 아니라 두 단계 이상을 벗어나야 하므로 정상적으로 작동하지 않는다.
이 경우 어떻게 해결하는가? 플래그? 예외?
예외는 로직을 제어하기 위한 수단이 아니므로 좋은 방법이 아니다.
가장 좋은 방법은 가능하면 중첩을 풀어서 1차원 루프를 만드는 것이다.
다음은 피해야 할 코드이다.
def search_nested_bad(array, desired_value):
coords = None
for i, row in enumerate(array):
for j, cell in enumerate(row):
if cell == desired_value:
coords = (i, j)
break
if coords is not None:
break
if coords is None:
raise ValueError(f"{desired_value} not found")
logger.info("value %r found at [%i, %i]", desired_value, *coords)
return coords
다음은 종료 플래그를 사용하지 않은 형태의 예이다.
def _iterate_array2d(array2d):
for i, row in enumerate(array2d):
for j, cell in enumerate(row):
yield (i, j), cell
def search_nested(array, desired_value):
try:
coord = next(
coord
for (coord, cell) in _iterate_array2d(array)
if cell == desired_value
)
except StopIteration:
raise ValueError("{desired_value} not found")
logger.debug("value %r found at [%i, %i]", desired_value, *coord)
return coord
최대한 중첩 루프를 제거하고 추상화하여 반복을 단순화한다.
2-2. 파이썬의 이터레이터 패턴
제너레이터는 이터러블 객체의 특별한 경우이며, 좋은 이터러블 객체를 만들면 보다 효율적이고 가독성 높은 코드를 작성할 수 있게 된다.
이터레이터는 __iter__() 와 __next__() 매직 메서드를 구현한 객체이다. 항상 이 두가지를 구현해야 하는 것은 아니다.
# 이터레이션 인터페이스
이터러블은 반복을 지원하는 객체로 크게 보면 아무 문제없이 for ... in ... 루프를 실행할 수 있다는 것을 뜻한다. 그러나 이터러블과 이터레이터는 다르다.
일반적으로 이터러블은 반복할 수 있는 어떤 것으로 실제 반복을 할 때는 이터레이터를 사용한다. 즉, __iter__ 매직 메서드를 통해 이터레이터를 반환하고, __next__ 매직 메서드를 통해 반복 로직을 구현하는 것이다.
이터레이터는 단지 내장 next() 함수 호출 시 일련의 값에 대해 한 번에 하나씩만 어떻게 생성하는지 알고 있는 객체이다. 이터레이터를 호출하지 않은 상태에서 다음 값을 요청 받기 전까지 그저 얼어있는 상태일 뿐이다. 이러한 의미에서 모든 제너레이터는 이터레이터이다.
| 파이썬 개념 | 매직 메서드 | 비고 |
| 이터러블(iterable) | __iter__ | 이터레이터와 함께 반복 로직을 만든다. 이것을 구현한 객체는 for ... in ... 구문에서 사용할 수 있다. |
| 이터레이터(iterator) | __next__ | 한 번에 하나씩 값을 생산하는 로직을 정의한다. 더 이상 생산할 값이 없을 경우는 StopIteration 예외를 발생시킨다. 내장 next() 함수를 사용해 하나씩 값을 읽어 올 수 있다. |
다음 코드는 이터러블하지 않은 이터레이터 객체의 예이다. 오직 한 번에 하나씩 값을 가져올 수만 있다.
class SequenceIterator:
def __init__(self, start=0, step=1):
self.current = start
self.step = step
def __next__(self):
value = self.current
self.current += self.step
return value
시퀀스에서 하나씩 값을 가져올 수는 있지만 반복할 수는 없다.
>>> si = SequenceIterator(1, 2)
>>> next(si)
1
>>> next(si)
3
>>> next(si)
5
>>> for _ in SequenceIterator(): pass
Traceback (most recent call last):
File "<input>", line 1, in <module>
TypeError: 'SequenceIterator' object is not iterable
이유는 당연히 __iter__() 메서드를 구현하지 않았기 때문이다.
# 이터러블이 가능한 시퀀스 객체
파이썬이 for 루프를 만나면 객체가 __iter__ 를 구현했는지 확인하고 있으면 그것을 사용한다. 그러나 없을 경우는 다른 대비 옵션을 가동한다.
객체가 시퀀스인 경우(즉 __getitem__() 과 __len__() 매직 메서드를 구현한 경우) 도 반복 가능하다. 이 경우 인터프리터는 IndexError 예외가 발생할 때까지 순서대로 값을 제공한다. IndexError 는 StopIteration 과 같이 중지를 알리는 역할을 한다.
class MappedRange:
"""특정 숫자 범위에 대해 맵으로 변환"""
def __init__(self, transformation, start, end):
self._transformation = transformation
self._wrapped = range(start, end)
def __getitem__(self, index):
value = self._wrapped.__getitem__(index)
result = self._transformation(value)
logger.debug("Index %d: %s", index, result)
return result
def __len__(self):
return len(self._wrapped)
위 예제는 오직 일반 for 루프를 사용해 반복 가능하다는 것을 보여주기 위한 것이다. __getitem__ 메서드에서는 객체가 반복하는 동안 어떤 값이 전달되었는지 확인하기 위해 로그를 출력한다.
>>> mr = MappedRange(abs, -10, 5)
>>> print(mr[0])
Index 0: 10
10
>>> print(mr[-1])
Index -1: 4
4
>>> print(list(mr))
Index 0: 10
Index 1: 9
Index 2: 8
Index 3: 7
Index 4: 6
Index 5: 5
Index 6: 4
Index 7: 3
Index 8: 2
Index 9: 1
Index 10: 0
Index 11: 1
Index 12: 2
Index 13: 3
Index 14: 4
[10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0, 1, 2, 3, 4]
이러한 방법이 있다는 것을 알아두는 것은 좋지만 객체가 __iter__ 를 구현하지 않았을 때 동작하는 대비책이다. 단순히 반복 가능한 객체를 만드는 것보다 적절한 시퀀스를 만들어 해결하는 것이 바람직하다.
객체가 시퀀스여서 우연히 반복이 가능할 수 있지만, 기본적으로 반복을 위한 객체를 디자인할 때는 __iter__ 매직 메서드를 구현하여 정식 이터러블 객체를 만들어야 한다.
3. 코루틴(coroutine)
코루틴은 cooperative routine 의 약자로 일반적으로 알고 있는 함수나 메서드 같은 서브루틴(sub-routine)이 메인루틴(main-routine) 과 종속관계를 가진 것과 다르게, 메인루틴과 대등한 관계로 협력하는 모습에서 코루틴이라고 불리게 되었다.
제너레이터를 코루틴으로도 활용할 수도 있다. 코루틴을 지원하기 위해 PEP-342에 추가된 기본 메서드는 다음과 같다.
- .close()
- .throw(ex_type[, ex_value[, ex_traceback]])
- .send(value)
3-1. 제너레이터 인터페이스
# close()
이 메서드를 호출하면 제너레이터에서 GeneratorExit 예외가 발생한다. 이 예외를 따로 처리하지 않으면 제너레이터가 더 이상 값을 생성하지 않으며 반복이 중지된다.
이 예외는 종료 상태를 지정하는데 사용될 수 있다. 코루틴이 일종의 자원 관리를 하는 경우 이 예외를 통해서 코루틴이 보유한 모든 자원을 해제할 수 있다. 일반적으로 컨텍스트 관리자를 사용하거나 finally 블록에 코드를 배치하는 것과 비슷하지만 이 예외를 사용하면 보다 명확하게 처리할 수 있다.
다음 예제는 코루틴을 사용하여 데이터베이스 연결을 유지한 상태에서 한 번에 모든 레코드를 읽는 대신에 특정 크기의 페이지를 스트리밍한다.
class DBHandler:
def __init__(self, db):
self.db = db
self.is_closed = False
def read_n_records(self, limit):
return [(i, f"row {i}") for i in range(limit)]
def close(self):
logger.debug("closing connection to database %r", self.db)
self.is_closed = True
def stream_db_records(db_handler):
try:
while True:
yield db_handler.read_n_records(10)
time.sleep(.1)
except GeneratorExit:
db_handler.close()
제너레이터를 호출할 때마다 데이터베이스 핸들러에서 얻은 10 개의 레코드를 반환하고, 명시적으로 반복을 끝내고 close() 를 호출하면 데이터베이스 연결도 함께 종료된다.
>>> streamer = stream_db_records(DBHandler("testdb"))
>>> next(streamer)
[((0, 'row 0'), (1, 'row 1'), ...]
>>> next(streamer)
[((0, 'row 0'), (1, 'row 1'), ...]
>>> streamer.close()
INFO: closing connection to database 'testdb'
제너레이터에서 작업을 종료할 때는 close() 메서드를 사용한다.
# throw(ex_type[, ex_value[, ex_traceback]])
이 메서드는 현재 제너레이터가 중단된 위치에서 예외를 던진다. 제너레이터가 예외를 처리했으면 해당 except 절에 있는 코드가 호출되고, 예외를 처리하지 않았으면 예외가 호출자에게 전파된다.
여기서는 코루틴이 예외를 처리했을 때와 그렇지 않을 때의 차이를 설명하기 위해 코드를 약간 수정했다.
class CustomException(Exception):
pass
def stream_data(db_handler):
while True:
try:
yield db_handler.read_n_records(10)
except CustomException as e:
logger.info("controlled error %r, continuing", e)
except Exception as e:
logger.info("unhandled error %r, stopping", e)
db_handler.close()
break
이제 CustomException 을 처리하고 있으며 이 예외가 발생한 경우 제너레이터는 INFO 레벨의 메시지를 기록한다. 그리고 다음 yield 구문으로 이동하여 데이터베이스에서 다시 데이터를 가져온다.
모든 예외를 처리하고 있지만 마지막 블록 (except Exception) 이 없으면 제너레이터가 중지된 행에서 예외가 호출자에게 전파되고 제너레이터는 중지된다.
>>> streamer = stream_data(DBHandler("testdb"))
>>> next(streamer)
[(0, 'row 0'), (1, 'row 1'), (2, 'row 2'), (3, 'row 3'), (4, 'row 4'), (5, 'row 5'), (6, 'row 6'), (7, 'row 7'), (8, 'row 8'), (9, 'row 9')]
>>> next(streamer)
[(0, 'row 0'), (1, 'row 1'), (2, 'row 2'), (3, 'row 3'), (4, 'row 4'), (5, 'row 5'), (6, 'row 6'), (7, 'row 7'), (8, 'row 8'), (9, 'row 9')]
>>> streamer.throw(CustomException)
controlled error CustomException(), continuing
>>> streamer.throw(RuntimeError)
unhandled error RuntimeError(), stopping
Traceback (most recent call last):
...
StopIteration
도메인에서 처리하고 있는 CustomException 예외를 받은 경우 제너레이터는 계속 진행된다. 그러나 그 외 예외는 Exception 으로 넘어가서 데이터베이스 연결을 종료하고 반복도 종료하게 된다. 마지막에 StopIteration 이 출력된 것에서 알 수 있듯이 제너레이터는 이제 더 이상 반복할 수 없다.
# send(value)
앞선 예제에서 제너레이터의 주요 기능은 고정된 수의 레코드를 읽는 것이다. 이제 읽어올 개수를 파라미터로 받아서 호출하도록 수정해보자. next() 함수는 이러한 옵션을 제공하지 않는다. 이럴 때 send() 메서드를 사용하면 된다.
def stream_db_records(db_handler):
retrieved_data = None
previous_page_size = 10
try:
while True:
page_size = yield retrieved_data
if page_size is None:
page_size = previous_page_size
previous_page_size = page_size
retrieved_data = db_handler.read_n_records(page_size)
except GeneratorExit:
db_handler.close()
이제 send() 메서드를 통해 인자 값을 전달할 수 있다. send() 메서드는 제너레이터와 코루틴을 구분하는 기준이 된다. send() 메서드를 사용했다는 것은 yield 키워드가 할당 구문의 오른쪽에 나오게 되고 인자 값을 받아서 다른 곳에 할당할 수 있음을 뜻한다.
코루틴에서는 일반적으로 다음과 같은 형태로 yield 키워드를 사용한다.
receive = yield produced
이 경우 yield 키워드는 두 가지 일을 한다. 하나는 produced 값을 호출자에게 보내고 그 곳에 멈추는 것이다. 호출자는 next() 메서드를 호출하여 다음 라운드가 되었을 때 값을 가져올 수 있다. 다른 하나는 거꾸로 호출자로부터 send() 메서드를 통해 전달된 produced 값을 받는 것이다. 이렇게 입력된 값은 receive 변수에 할당된다.
코루틴에 값을 전송하는 것은 yield 구문이 멈춘 상태에서만 가능하다. 그렇게 되려면 일단 코루틴을 해당 상태까지 이동시켜야 한다. 코루틴이 해당 상태로 이동하는 유일한 방법은 next() 를 호출하는 것이다. 즉 코루틴에게 무엇인가를 보내기 전에 next() 메서드를 적어도 한 번은 호출해야 한다는 것을 의미한다. 그렇지 않으면 다음과 같은 예외가 발생한다.
>>> c = coro()
>>> c.send(1)
Traceback (most recent call last):
...
TypeError: can't send non-None value to a just-start generator
코루틴에서 send() 메서드를 호출하려면 항상 next()를 먼저 호출해야 한다.
제너레이터에서 처음 next() 를 호출하면 yield 를 포함하는 위치까지 이동한다. 그리고 현재 상태의 변수 값을 반환하고 거기에 멈춘다. 변수의 초기 값이 None 이므로 처음 next() 를 호출하면 None 을 반환한다. 여기에서 두 가지의 옵션이 있다. 그냥 next() 를 호출하면 기본값인 10을 사용하여 평소처럼 이후 작업이 계속된다. next() 는 send(None) 과 같기 때문에 if page_size is None 에서 기본 값을 사용하도록 설정된다.
반면에 send(<value>) 를 통해 명시적인 값을 제공하면 yield 문의 반환 값으로 page_size 변수에 설정된다. 이제 기본 값이 아닌 사용자가 지정한 값이 page_size 로 설정되고 해당 크기만큼 데이터베이스에서 레코드를 읽어오게 된다.
이어지는 호출에 대해서도 같은 로직이 적용된다. 중요한 것은 이제 아무 때나 페이지 크기를 지정할 수 있다는 점이다.
동작 원리를 이해했으므로 보다 간결하고 깔끔하게 수정할 수 있다.
def stream_db_records(db_handler):
retrieved_data = None
page_size = 10
try:
while True:
page_size = (yield retrieved_data) or page_size
retrieved_data = db_handler.read_n_records(page_size)
except GeneratorExit:
db_handler.close()
yield 주변의 괄호는 해당 문장이 함수를 호출하는 것처럼 사용되고 page_size 와 비교할 것이라는 점을 명확히 한다.
코드가 보다 간결해졌지만 여전히 send() 전에 next() 를 먼저 호출해야 한다. 그렇지 않으면 TypeError 가 발생한다. 이를 해결하기 위한 데코레이터를 사용하면 아래와 같다.
def prepare_coroutine(coroutine):
def wrapped(*args, **kwargs):
advanced_coroutine = coroutine(*args, **kwargs)
next(advanced_coroutine)
return advanced_coroutine
return wrapped
@prepare_coroutine
def auto_stream_db_records(db_handler):
retrieved_data = None
page_size = 10
try:
while True:
page_size = (yield retrieved_data) or page_size
retrieved_data = db_handler.read_n_records(page_size)
except GeneratorExit:
db_handler.close()
이제 next() 를 호출하지 않고도 코루틴을 바로 사용할 수 있다.
>>> streamer = auto_stream_db_records(DBHandler("testdb"))
>>> len(streamer.send(5))
5
3-2. 코루틴 고급 주제
# 코루틴에서 값 반환하기
반복이란 StopIteration 예외가 발생할때까지 next() 메서드를 계속해서 호출하는 메커니즘을 말한다.
지금까지는 한 번에 하나씩 값을 생성하는 제너레이터의 특성을 확인했으며 일반적으로 for 루프의 모든 단계에서 생성되는 각 값에 대해서만 신경쓴다. 코루틴은 기술적으로는 제너레이터지만 반복을 염두에 두고 만든 것이 아니라 나중에 코드가 실행될 때까지 실행을 멈추는 것을 목표로 한다.
코루틴은 일반적으로 반복보다는 상태를 중단하는데 초점을 맞추고 있다. 오히려 반복을 목적으로 코루틴을 만드는 것이 이상한 경우일 것이다. 파이썬에서는 코루틴은 기술적으로 제너레이터에 기반을 두고 있기 때문에 이 둘의 개념이 섞일 수 있다.
제너레이터는 일반 함수가 아니므로 value = generator() 라고 하는 것은 제너레이터 객체를 만드는 것 외에는 아무것도 하지 않는다. 제너레이터가 값을 반환하게 하려면 반복을 해야 한다.
제너레이터에서 값을 반환하면 반복이 즉시 중단된다. 더 이상 반복할 수 없다. 본래의 의미 체계를 유지하기 위해 StopIteration 예외가 발생해도 예외 객체 내에 반환 값이 저장되어 있다. 예외에서 해당 값을 처리하는 것은 호출자의 책임이다.
다음 예제에서는 제너레이터를 사용해 두 개의 값을 생성하고 세 번째 값을 반환한다. 마지막 return 되는 값을 구하기 위해 예외를 처리하는 방법과 예외에서 어떻게 값을 구하는지 유의해서 살펴보자.
>>> def generator():
yield 1
yield 2
return 3
>>> value = generator()
>>> next(value)
1
>>> next(value)
2
>>> try:
next(value)
except StopIteration as e:
print(">>>>> return value ", e.value)
>>>>> return value 3
3-3. 작은 코루틴에 위임하기 - yield from 구문
코루틴이 값을 반환할 수 있지만 반환 기능 자체는 언어에서 지원해주지 않으면 귀찮은 부분이 있다. 이 점을 개선한 구문이 yield from 구문이다. 이전에 제너레이터가 값을 반환하는 것은 가능하지만 value = generator() 같은 동작이 작동하지 않았다. 이제 value = yield from generator() 와 같이 작성하면 가능해진다.
# 가장 간단한 yield from 사용 예
가장 간단한 형태의 yield from 구문은 제너레이터 체인에서 살펴볼 수 있다. 제너레이터 체인은 여러 제너레이터를 하나의 제너레이터로 합치는 기능을 하는데 중첩된 for 루프를 사용해 하나씩 모으는 대신에 서브 제너레이터 값을 한 번에 수집할 수 있게 해준다.
대표적인 예로 표준 라이브러리인 itertools.chain() 과 비슷한 함수를 만들어본다.
def chain(*iterables):
for it in iterables:
for value in it:
yield value
여러 이터러블을 받아서 모두 이동한다. 모두 이터러블이므로 for ... in 구문을 지원하므로 개별 값을 구하려면 중첩 루프를 사용하면 된다. 이렇게 하면 리스트를 튜플과 비교하는 것처럼 직접 비교가 어려운 자료형에 대해서도 한 번에 처리할 수 있으므로 편리하다.
여기서 yield from 구문을 사용하면 서브 제너레이터에서 직접 값을 생성할 수 있으므로 중첩 루프를 피할 수 있다.
def chain(*iterables):
for it in iterables:
yield from it
앞의 제너레이터와 결과는 모두 동일하다.
>>> list(chain("hello", ["world"], ("tuple", " of ", "values.")))
['h', 'e', 'l', 'l', 'o', 'world', 'tuple', ' of ', 'values.']
yield from 구문은 어떤 이터러블에 대해서도 동작하며 이것을 사용하면 마치 최상위 제너레이터가 직접 값을 yield 한 것과 같은 효과를 나타낸다.
yield from 은 어떤 형태의 이터러블에서도 동작하므로 제너레이터 표현식도 마찬가지이다. 이제 yield from 구문을 활용해 입력된 파라미터의 모든 제곱지수를 만드는 제너레이터를 만들어 보자. 예를 들어 all_powers(2, 3) 은 2^0, 2^1, 2^2, 2^3 을 만드는 것이다.
def all_powers(n, pow):
yield from (n ** i for i in range(pow + 1))
이렇게 하면 기존의 서브 제너레이터에서 for 문을 사용해 값을 생산하는 대신 한 줄로 직접 값을 생산할 수 있으므로 편리하지만, 이것만으로 yield from 을 언어에서 지원해야만 하는 이유였다고 보기는 어렵다.
# 서브 제너레이터에서 반환된 값 구하기
다음 예제는 수열을 생산하는 두 개의 중첩된 제너레이터를 호출한다. 각각의 제너레이터는 값을 반환하는데 최상위 제너레이터는 쉽게 반환 값을 확인할 수 있다. 바로 yield from 구문을 사용했기 때문이다.
def sequence(name, start, end):
logger.info("%s started at %i", name, start)
yield from range(start, end)
logger.info("%s finished at %i", name, end)
return end
def main():
step1 = yield from sequence("first", 0, 5)
step2 = yield from sequence("second", step1, 10)
return step1 + step2
위 코드를 실행하면 다음과 같이 동작한다.
>>> g = main()
>>> next(g)
INFO:first started at 0
0
>>> next(g)
1
>>> next(g)
2
>>> next(g)
3
>>> next(g)
4
>>> next(g)
INFO:first finished at 5
INFO:second started at 5
5
>>> next(g)
6
>>> next(g)
7
>>> next(g)
8
>>> next(g)
9
>>> next(g)
INFO:second finished at 10
Traceback (most recent call last):
...
StopIteration: 15
main 제너레이터의 첫 번째 행은 내부 제너레이터로 위임하여 생산된 값을 가져온다. sequence() 제너레이터 종료 시의 반환 값을 step1 으로 받아와서 다음 sequence() 제너레이터에 전달한다.
두 번째 제너레이터 역시 종료 시 값(10)을 반환하고 main 제너레이터는 이 두 결과의 합(5+10=15) 을 반환하며 이 값은 StopIteration 에 포함되어있다.
* 서브 제너레이터에 return 문장이 없을 경우 yield from 반환 값은 None 이다.
yield from 을 사용하면 코루틴의 종료 시 최종 반환 값을 구할 수 있다.
# 서브 제너레이터와 데이터 송수신하기
앞의 예제에서 최상위 main 제너레이터는 그대로 유지하고, 값을 수신하고 예외를 처리할 내부 제너레이터인 sequence 함수를 아래와 같이 수정한다.
def sequence(name, start, end):
value = start
logger.info("%s started at %i", name, value)
while value < end:
try:
received = yield value
logger.info("%s received %r", name, received)
value += 1
except CustomException as e:
logger.info("%s is handling %s", name, e)
received = yield "OK"
return end
def main():
step1 = yield from sequence("first", 0, 5)
step2 = yield from sequence("second", step1, 10)
return step1 + step2
이제 main 코루틴을 반복하는 것 뿐 아니라 내부 sequence 제너레이터에서 어떻게 처리하는지 확인하기 위해 값을 전달하거나 예외를 던져본다.
>>> g = main()
>>> next(g)
INFO:first started at 0
0
>>> next(g)
INFO:first received None
1
>>> g.send("첫 번째 제너레이터를 위한 인자 값")
INFO:first received '첫 번째 제너레이터를 위한 인자 값'
2
>>> g.throw(CustomException("처리 가능한 예외"))
INFO:first is handling 처리 가능한 예외
'OK'
>>> next(g)
2
>>> next(g)
INFO:first received None
3
>>> next(g)
INFO:first received None
4
>>> next(g)
INFO:first received None
INFO:second started at 5
5
>>> g.throw(CustomException("두 번째 처리 가능한 예외"))
INFO:second is handling 두 번째 처리 가능한 예외
'OK'
sequence 서브 제너레이터에 값을 보내지 않고 오직 main 제너레이터에 값을 보냈다. 실제 값을 받은 것은 내부 제너레이터이다. 명시적으로 sequence 에 데이터를 보낸 적은 없지만 실질적으로는 yield from 을 통해 sequence 에 데이터를 전달한 셈이다.
main 코루틴은 내부적으로 두 개의 다른 코루틴을 호출하여 값을 생산하며 특정 시점에서 보면 둘 중에 하나는 멈춰져 있는 상태다. 로그를 통해 첫 번째 코루틴이 멈춰진 상태에서 데이터를 전송해도 첫 번째 코루틴 인스턴스가 값을 받는다는 것을 알 수 있다. 예외를 던질 때도 마찬가지다. 첫 번째 코루틴이 끝나면 step1 변수에 값을 반환하고, 그 값을 두 번째 코루틴에 입력으로 전달한다. 두 번째 코루틴도 첫 번째와 동일하게 send() 와 throw() 에 대해 동일한 작업을 한다.
각 코루틴이 생성하는 값에 대해서도 마찬가지다. 특정 단계에서 send() 를 호출했을 때 생성하는 값은 사실 현재 main 제너레이터가 멈춰 있던 서브 코루틴에서 생산한 값이다. 처리 가능한 CustomException 예외를 던지면 sequence 코루틴에서 OK 를 생산하며 호출자 코루틴인 main 에 전파한다.
4. 비동기 프로그래밍
지금까지 살펴본 것들을 활용하여 비동기 프로그래밍을 만들 수 있다. 즉 여러 코루틴이 특정 순서로 동작하도록 스케줄링 할 수 있으며, 일시 정지된 yield from 코루틴과 통신할 수 있다.
이러한 기능을 통해 얻을 수 있는 가장 큰 장점은 논블로킹(non-blocking) 방식으로 병렬 I/O 작업을 할 수 있다는 것이다. 보통 서드파티 라이브러리에서 구현한 저수준 제너레이터를 활용한다. 코루틴이 정지된 동안 프로그램은 다른 작업을 수행하고, 프로그램은 yield from 문장에 의해 중단되기도 하고 생산된 값을 받기도 하며 제어권을 주고 받는다.
코루틴과 제너레이터가 기술적으로는 동일하지만 의미적으로는 다르다. 효율적인 반복을 원할 때는 제너레이터를 사용하고, 논블로킹 I/O 작업을 원할 때는 코루틴을 사용한다.
result = yield from iterable_or_awaitable()
위와 같은 코드가 있다고 가정했을 때, iterable_or_awaitable 이 반환하는 것이 명확하지 않다. 코루틴과 제너레이터의 사용 목적이 분명히 다르므로, 기술적으로 동일하다고 해서 불분명한 반환 값을 사용하는 것은 좋지 않다.
이러한 이유 때문에 파이썬의 타이핑 시스템이 확장되었다. 파이썬 3.5 이상부터 코루틴이라는 새로운 타입이 추가되었다.
또한 새로운 구문으로 await 와 async def 또한 추가되었다. await 는 yield from 을 대신하기 위한 용도로 awaitable 객체에 대해서만 동작한다. 코루틴은 awaitable 이다. awaitable 인터페이스를 따르지 않는 객체에 await 를 호출하면 예외가 발생한다. async def 는 @coroutine 데코레이터를 대신하여 코루틴을 정의하는 새로운 방법이다. 이것은 호출 시 실제로 객체를 만들어 코루틴 인스턴스를 반환한다.
파이썬에서 비동기 프로그래밍을 한다는 것은 일련의 코루틴을 관리하는 이벤트 루프가 있다는 뜻이다. 일련의 코루틴들은 이벤트 루프에 속하며, 이벤트 루프의 스케줄링 메커니즘에 따라 호출된다. 각각의 코루틴이 실행되면 사용자가 작성한 내부 코드가 실행되고 다시 이벤트 루프에 제어권을 반납하려면 await <coroutine> 을 호출하면 된다. await <coroutine> 는 yield 처럼 호출자(이벤트 루프)에 제어권을 넘겨줌으로써 이벤트 루프가 작업을 비동기적으로 관리할 수 있게 해준다.
'python' 카테고리의 다른 글
| 파이썬 클린 코드 - 9장 (일반적인 디자인 패턴) (0) | 2022.04.13 |
|---|---|
| 파이썬 클린 코드 - 8장 (단위 테스트와 리팩토링) (0) | 2022.03.16 |
| 파이썬 클린 코드 - 6장 (디스크립터로 더 멋진 객체 만들기) (0) | 2022.03.03 |
| 파이썬 클린 코드 - 5장 (데코레이터를 사용한 코드 개선) (0) | 2022.02.18 |
| 파이썬 클린 코드 - 4장 (SOLID 원칙) (0) | 2022.02.09 |