파이썬 클린코드 | 7장 제너레이터, 이터레이터 및 비동기 프로그래밍

제너레이터 만들기

제너레이터 개요

class PurchaseStats:
	def __init__(self, purchases):
		self.purchases = iter(purchases)
		self.min_price: float = None
		self.max_price: float = None
		self._total_purchases_price: float = None
		self._total_purchases = 0
		self._initialize()

	def _initialize(self):
		try:
			first_value = next(self.purchases)
		except StopIteration:
			raise ValueError("더이상 값이 없음")	

		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_value:
			self.min_value = new_value

	def _update_max(self, new_value: float):
		if new_value > self.max_price:
			self.max_value = 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)

>>> load_purchases("file")
<generator object load_purchases at 0X....> # 이터러블한 객체라고 부른다.

제너레이터 표현식

>>> [x**2 for x in range(10)]
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

>>> (x**2 for x in range(10))
<generator object <genexpr> at 0x....>

>>> sum(x**2 for x in range(10))
285

이상적인 반복

관용적인 반복 코드

>>> 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

>>> seq = NumberSequence()
>>> seq.next()
0
>>> seq.next()
1

>>> seq2 = NumberSequence(10)
>>> seq2.next()
10
>>> seq2.next()
11

>>> list(zip(NumberSequence(), "abcdef"))
Traceback (most recent call last):
	File "...", line 1, in <module>
TypeError: zip argument #1 must support iteration 

class SequenceOfNumber:
	def __init__(self, start=0):
		self.current = start

	def __next__(self):
		current = self.current
		self.current += 1
		return current

	def __iter__(self):
		return self

>>> list(zip(SequenceOfNumber(), "abcdef"))
[(0, "a"), (1, "b"), (2, "c"), (3, "d"), (4, "e"), (5, "f")]
>>> seq = SequenceOfNumber(100)
>>> next(seq)
100
>>> next(seq)
101

next() 함수

>>> word = iter("hello")
>>> next(word)
"h"
>>> next(word)
"e"
...
>>> next(word)
"o"
>>> next(word)
Traceback (most recent call last):
	File "...", line 1, in <module>
StopIteration

>>> next(word, "default value")
"default value"

제너레이터 사용하기

def sequence(start=0):
	while True:
		yield start
		start += 1

>>> 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

# 위 예제 중 구매 이력에서 지표를 계산하는 과정
# 만약 특정 기준을 넘은 값에 대해서만 연산을 하고 싶다면?

def process(self):
	for purchase in self.purchases:
		if purchase > 1000.0:
			...

이터레이터를 사용한 코드 간소화

def process_purchases(purchases):
	min_, max_, avg = itertools.tee(purchases, 3)
	return min(min_), max(max_), median(avg)

def search_nexted_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} 값을 찾을 수 없음")

	logger.info("[%i, %i]에서 값 %r 발견", *coords, desired_value)
	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 ValueErro("{desired_value} 값을 찾을 수 없음")

	logger.info("[%i, %i]에서 값 %r 발견", *coords, desired_value)
	return coords

파이썬의 이터레이터 패턴

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.info("Index %d: %s", index, result)
		return result

	def __len__(self):
		return len(self._wrapped)

>>> mr = MappedRange(abs, -10, 5)
>>> mr[0]
Index 0: 10
10
>>> mr[-1]
Index -1: 4
4
>>> list(mr)
Index 0: 10
Index 1: 9
Index 2: 8
...
Index 9: 1
Index 10: 0
Index 11: 1
Index 12: 2
...
Index 14: 4
[10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0, 1, 2, 3, 4]

코루틴

제너레이터 인터페이스의 메서드

close

def stream_db_records(db_handler):
	try:
		while True:
			yield db_handler.read_n_records(10)
	except GeneratorExit:
		db_handler.close()

>>> streamer = stream_db_records(DBHandler("testdb"))
>>> next(streamer)
[(0, 'row 0'), (1, 'row 1'), (2, 'row 2'), (3, 'row 3')....]
>>> next(streamer)
[(0, 'row 0'), (1, 'row 1'), (2, 'row 2'), (3, 'row 3')....]
>>> streamer.close()
INFO:....: 'testdb' 데이터 베이스 연결 종료

throw(ex_type[, ex_value[, ex_traceback]])

class CustomException(Exception):
	"""처리하려는 에러 유형"""

	def stream_data(db_handler):
		while True:
			try:
				yield db_handler.read_n_records(10)
			except CustomException as e:
				logger.warning("%r 에러 발생 후 계속 진행", e)
			except Exception as e:
				logger.error("%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')....]
>>> next(streamer)
[(0, 'row 0'), (1, 'row 1'), (2, 'row 2'), (3, 'row 3')....]
>>> streamer.throw(CustomerException)
WARNING: CustomException() 에러 발생 후 계속 진행
[(0, 'row 0'), (1, 'row 1'), (2, 'row 2'), (3, 'row 3')....]
>>> streamer.throw(RuntimeError)
ERROR: RuntimeError() 발생 후 중단
INFO: 'testdb' 데이터베이스 연결 종료
Trackback (most recent call last):
	....
StopIteration

send(value)

def stream_db_record(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()

• produced 값을 호출자에게 보내고 그 곳을 멈춘다. 호출자는 next() 메서드를 호출해 다음 라운드에서 값을 가져올 수 있다.

• 호출자로부터 send() 메서드를 통해 전달된 produced 값을 받는 것이다. 이렇게 입력된 값은 receive 변수에 할당된다.

>>> def coro():
			y = yield
....
....
>>> c = coro()
>>> c.send(1)
TypeError: can't send non-None Value to a just-started generator

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()

• next()를 호출하면 기본값인 10을 반환. 평소처럼 작업이 계속된다. 여기서 next()는 send(None) 과 같다.

• send(<value>)를 통해 명시적인 값을 제공하면 yield 문의 반환 값으로 page_size의 변수에 설정된다. 기본 값이 아닌 사용자가 지정한 값이 page_size로 설정되고, 해당 크기만큼 데이터 베이스에서 읽어온다. 이어지는 호출에 대해서도 같은 값이 적용된다

코루틴 고급 주제

코루틴에서 값 반환하기

>>> def generator():
			yield 1
			yield 2
			yield 3
....
....
>>> value = generator()
>>> next(value)
1
>>> next(value)
2
>>> try:
			next(value)
		except StopIteration as e:
			print(f"최종 반환값: {e.value}")
최종 반환값 3

작은 코루틴에 위임하기 - yield from 구문

가장 간단한 yield from 사용 예

def chain(*iterables):
	for it in iterables:
		for value in it:
			yield value

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"]

서브 제너레이터에서 반환된 값 구하기

서브 제너레이터와 데이터 송수신하기

비동기 프로그래밍

비동기 매직 메서드

비동기 컨텍스트 관리자

다른 매직 메서드

비동기 반복

비동기 제너레이터