제너레이터 만들기
제너레이터는 한 번에 하나씩 구성요소를 반환해주는 이터러블을 생성해주는 객체이다. 제너레이터는 필요할 때마다 요소를 하나씩만 가져오는 방식으로 메모리를 절약한다.
제너레이터 개요
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})"
)이 객체는 모든 구매 정보 purchases를 받아서 필요한 계산을 한다.
다음 코드는 모든 정보를 로드해 반환해주는 함수이다.
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이 코드는 정상적인 결과를 반환한다. 하지만 성능에 문제가 있다. 파일에 많은 데이터가 있다면 로드시에 시간이 오래 걸리고 메모리에 담지 못할 수도 있다.
PurchasesStats 객체에서는 한 번에 하나의 데이터만 사용하므로 한번에 모든 데이터를 읽어와 메모리에 보관할 이유가 없다.
이에대한 해결책은 제너레이터를 만드는 것이다. 제너레이터는 파일의 전체 내용을 리스트에보관하는 대신 필요한 값만 그때그때 가져온다.
def load_purchases(filename):
with open(filename) as f:
for line in f:
*_, price_raw = line.partition(",")
yield float(price_raw)이렇게 수정하면 메모리 사용량을 줄일 수 있다. 결과 리스트를 만들 필요도 없으며 return문도 사라졌다.
이 load_purchases함수를 제너레이터라고 부른다. 파이썬에서는 어떤 함수라도 yield 키워드를 사용하면 제너레이터가 된다. yield가 포함된 함수를 호출하면 제너레이터의 인스턴스를 만든다.
모든 제너레이터 객체는 이터러블이다. 이터러블은 for 루프와 함께 사용할 수 있다. 이터러블을 사용하면 for루프의 다형성을 보장하는 강력한 추상화가 가능하다.
제너레이터 표현식
제너레이터를 사용하면 많은 메모리를 절약할 수 있다. 제너레이터는 이터레이터이므로 리스트, 튜플, 세트처럼 많은 메모리를 필요로 하는 이터러블이나 컨테이너의 대안이 될 수 있다.
컴프리헨션(comprehension)에 의해 정의될 수 있는 리스트, 세트, 딕셔너리처럼 제너레이터도 제너레이터 표현식으로 정의할 수 있다.
리스트 컴프리헨션에서 대괄호를 괄호로 교체하면 표현식의 결과로부터 제너레이터가 생성된다. 제너레이터 표현식은 sum()이나 max()같이 이터러블 연산이 가능한 함수에 직접 전달할 수도 있다.
>>> (x**2 for x in range(10))
<generator object <genexpr> at 0x...>이상적인 반복
관용적인 반복 코드
파이썬에서 이터레이터 객체를 만들기 위해서는 __iter__()매직 메서드를 구현하여 객체가 반복 가능하게 만들어야 한다. 또한 __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() 내장함수는 이터러블을 다음 요소로 이동시키고 기존의 값을 반환한다.
이터레이터가 더 이상의 값을 가지고 있지 않다면 StopIteration 예외가 발생한다.
이 문제의 해결 방법에는 StopIteration 예외를 캐치하는 것 이외에도 next() 함수의 두 번째 파라미터에 기본 값을 제공할 수도 있다. 이 값을 제공하면 StopIteration예외를 발생시키는 대신 기본 값을 반환한다.
제너레이터 사용하기
제너레이터를 사용하면 클래스를 만드는 대신 필요한 값을 yield하는 함수를 만들면 된다.
def sequence(start=0):
while True:
yield start
start += 1제너레이터 함수가 호출되면 yield문장을 만나기 전까지 실행된다. 그리고 값을 생성하고 그 자리에서 멈춘다.
itertools
처음 예제에서 특정 기준을 넘은 값에 대해서만 연산을 하고자 할 경우 반복문 안에 조건문을 추가하면 된다.
def process(self)
for purchase in self.purchases:
if purchase > 1000.0:
...이것은 pythonic하지 않을 뿐만 아니라 너무 엄격하다. 이 객체의 고유 책임은 요소에 대해 지표 값을 계산하고 출력하는 것뿐이다.
이 객체는 클라이언트의 요구로부터 독립되어야 한다.
from itertools import islice
purchases = islice(filter(lambda p: p > 1000.0, purchases), 10)이런 식으로 필터링을 해도 메모리의 손해는 없다. 모든 것이 제너레이터이므로 요소 하나씩 평가된다.
이터레이터를 사용한 코드 간소화
여러번 반복하기
itertools모듈을 사용해 처음의 예제를 간소화할 수 있다.
itertools.tee는 원래의 이터러블을 여러 개의 새로운 이터러블로 복제한다.
def process_purchases(purchases):
min_, max_, avg = itertools.tee(purchases, 3)
return min(min_), max(max_), median(avg)중첩루프
1차원 이상을 반복하여 값을 찾아야 할 수 있다. 중첩 루프를 사용할 시 값을 찾으면 순환을 멈추고 break키워드를 호출해야 하는데 이 경우 두 단계 이상을 버어나야 하므로 정상 동작하지 않는다.
이 경우 플래그나 예외를 사용할 수 있지만 이것들은 코드를 나쁘게 만든다.
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가장 좋은 방법은 중첩을 풀어 1차원 루프로 만드는 것이다.
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("[%i, %i]에서 값 %r 찾음",*coord, desired_value)
return coord파이썬의 이터레이터 패턴
이터레이션 인터페이스
이터러블은 반복할 수 있는 것으로 실제 반복을 할 때에는 이터레이터를 사용한다. 즉 __iter__매직 메서드를 사용하여 이터레이터를 반환하고 __next__ 매직 메서드를 통해 반복 로직을 구현한다.
| 파이썬 개념 | 매직 메서드 | 비고 |
|---|---|---|
| 이터러블 | __iter__ | 이터레이터와 함께 반복 로직을 만든다.for in 구문에서 사용할 수 있다. |
| 이터레이터 | __next__ | 한 번에 하나의 값을 생성하는 로직을 정의한다. 더 이상 생성할 값이 없을 경우 StopIteration 예외를 발생시킨다. next()를 사용할 수 있다. |
이터러블이 가능한 시퀀스 객체
__iter__() 매직 메서드를 구현한 객체는 for 루프에서 사용할 수 있다. 이것은 큰 특징이지만 꼭 이런 형태여만 반복이 가능한 것은 아니다. 파이썬이 for 루프를 만나면 객체가 __iter__을 구현했는지 확인하여 있으면 그것을 사용한다. 없을 경우에는 다른 대비 옵션을 사용한다.
객체가 시퀀스인 경우도 반복이 가능하다. 이 경우 인터프리터는 IndexError 예외가 발생할 때까지 순서대로 값을 제공한다.
코루틴
제너레이터 인터페이스 메서드
close()
이 메서드를 호출하면 제너레이터에서 GeneratorExit 예외가 발생한다. 이 예외를 따로 처리하지 않으면 제너레이터가 더 이상 값을 생성하지 않으며 반복이 중지된다.
이 예외는 종료 상태를 지정하는데 사용할 수 있다. 코루틴이 자원관리를 하는 경우 이 예외를 통해 코루틴이 보유한 모든 자원을 해제할 수 있다. 컨텍스트 관리자, finally와 비슷하지만 이 예외를 사용하면 보다 명확하게 처리할 수 있다.
def stream_db_records(db_handler):
try:
while True:
yield db_handler.read_n_records(10)
except GeneratorExit:
db_handler.close()위 예제는 코루틴을 사용하여 데이터베이터베이스에서 특정 크기의 페이지를 스트리밍한다.
제너레이터에서 작업을 종료할 때는
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("처리 가능한 에러 %r, 계속 진행", e)
except Exception as e:
logger.info("처리 불가능한 에러 %r, 중단", e)
db_handler.close()
break>>> 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)
처리 가능한 에러 CustomException(), 계속 진행
>>> streamer.throw(RuntimeError)
처리 불가능한 에러 RuntimeError(), 중단
Traceback (most recent call last):CustomException 예외가 발생한 경우 제너레이터는 INFO 레벨의 메세지를 기록하고 다음 yield 구문으로 이동하여 데이터베이스에서 데이터를 가져온다.
마지막 블록 (except Exception)가 없을 경우 제너레이터가 중지된 행에서 예외가 호출자에게 전파되고 제너레이터는 중지된다.
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()코루틴에서는 일반적으로 다음과 같은 형태로 yield 키워드를 사용한다.
receive = yield produced이 경우 yield 키워드는 두 가지 일을 한다. 하나는 produced 값을 호출자에게 보내는 것이고 다른 하나는 호출자로부터 send() 메서드를 통해 전달된 produced값을 받는 것이다.
코루틴에 값을 전송하는 것은 yield 구문이 멈춘 상태에서만 가능하다. 즉 코루틴에게 값을 보내기 전에 next() 메서드를 최소 한 번은 호출해야 한다.
코루틴에서 send() 메서드를 호출하려면 항상 next()를 먼저 호출해야 한다.
위 코드에서 send()를 사용해 값을 제공하면 yield 문의 반환 값으로 page_size변수에 할당된다. 이제 기본값인 previous_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()코드가 보다 간결해졌지만 여전히 send() 전에 next() 를 먼저 호출해야 한다. 그렇지 않으면 TypeError 가 발생한다. 이를 데코레이터를 사용해 해결할 수 있다.
send() 전에 next()를 호출하지 않고 코루틴을 생성하자마자 바로 사용할 수 있게 하기 위해 데코레이터를 사용한다.
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()작은 코루틴에 위임하기 - yield from 구문
코루틴이 값을 반환할 수 있도록 지원하는 구문이 yield from이다.
가장 간단한 yield from 사용 예
def chain(*iterables):
for it in iterables:
for value in it:
yield value여러 이터러블을 받아 개별 값들을 가져오는 제너레이터이다. 여기서 yield from 구문을 사용해 코드를 단순화시킬 수 있다.
def chain(*iterables):
for it in iterables:
yield from it서브 제너레이터에서 반환된 값 구하기
def sequence(name, start, end):
logger.info("%s 제너레이터 %i에서 시작", name, start)
yield from range(start, end)
logger.info("%s 제너레이터 %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:generators_yieldfrom_2:first 제너레이터 0에서 시작
0
>>> next(g)
1
>>> next(g)
2
>>> next(g)
3
>>> next(g)
4
>>> next(g)
INFO:generators_yieldfrom_2:first 제너레이터 5에서 종료
INFO:generators_yieldfrom_2:second 제너레이터 5에서 시작
5
>>> next(g)
6
>>> next(g)
7
>>> next(g)
8
>>> next(g)
9
>>> next(g)
INFO:generators_yieldfrom_2:second 제너레이터 10에서 종료
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
StopIteration: 15main 제너레이터는 내부 제너레이터에서 생산된 값을 가져온다.
그리고 sequence()제너레이터 종료 시의 반환 값을 다음 sequence()제너레이터에 전달한다.
yield from을 사용하면 코루틴의 종료 시 최종 반환값을 구할 수 있다.
서브 제너레이터와 데이터 송수신하기
앞의 예제에서 최상위 main 제너레이터는 그대로 유지하고 값을 수신하고 예외를 처리할 내부 제너레이터 sequence함수를 수정한다.
def sequence(name, start, end):
value = start
logger.info("%s 제너레이터 %i에서 시작", name, value)
while value < end:
try:
received = yield value
logger.info("%s 제너레이터 %r 값 수신", name, received)
value += 1
except CustomException as e:
logger.info("%s 제너레이터 %s 에러 처리", name, e)
received = yield "OK"
return end>>> g = main()
>>> next(g)
INFO: first 제너레이터 0에서 시작
0
>>> next(g)
INFO: first 제너레이터 None 값 수신
1
>>> g.send("첫 번째 제너레이터를 위한 인자 값")
INFO: first 제너레이터 '첫 번째 제너레이터를 위한 인자 값' 값 수신
2
>>> g.throw(CustomException("처리 가능한 예외 던지기"))
INFO: first 제너레이터 처리 가능한 예외 던지기 에러 처리
'OK'
>>> next(g)
INFO: first 제너레이터 None 값 수신
2
>>> next(g)
INFO: first 제너레이터 None 값 수신
3
>>> next(g)
INFO: first 제너레이터 None 값 수신
4
>>> next(g)
INFO: first 제너레이터 None 값 수신
INFO:second 제너레이터 5에서 시작
5
>>> g.throw(CustomException("두 번째 제너레이터를 향한 예외 던지기"))
INFO: second 제너레이터 두 번째 제너레이터를 향한 예외 던지기 예외 처리
'OK'이 예제에서 오직 main 제너레이터에만 값을 보냈지만 실제 값은 sequence 서브 제너레이터가 받는다. 명시적으로 sequence에 데이터를 보낸 적은 없지만 yield from을 통해 sequence에 데이터를 전달한다.
비동기 프로그래밍
비동기 프로그래밍의 장점은 논블로킹 방식으로 병렬I/O 작업을 할 수 있다는 것이다.
제너레이터와 코루틴은 문법적, 기술적으로는 동일하지만 의미적으로는 다르다. 효율적 반복을 원할 때는 제너레이터, 논블로킹I/O 작업을 원할 때는 코루틴을 사용한다.
제너레이터와 코루틴의 혼합으로 발생하는 런타임 오류를 줄이기 위해 python3.6 이후로 코루틴이라는 새로운 타입이 추가되었다.
새로운 구문으로 await와 async def또한 추가되었다. await는 yield from을 대신하기 위한 용도로 awaitable 객체에 대해서만 동작한다. async def는 @coroutine 데코레이터를 대신하여 코루틴을 정의하는 새로운 방법이다.