파이썬 클린코드를 읽으며 정리한 내용입니다.
def load_purchases(filename):
with open(filename) as f:
for line in f:
*_, price_raw = line.partition(",")
yield float(price_raw)
# list comprehension
[x**2 for x in range(10)]
>>> [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
# generator expression(comprehension)
(x**2 for x in range(10))
>>> <generator object <genexpr> at 0x7f92b8447e40>
# 시작 값을 입력하면 무한 시퀀스를 만드는 클래스
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
__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()
내장함수를 사용할 수 있다.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()함수
word = iter("hello")
next(word)
>>> 'h'
next(word)
>>> 'e'
next(word)
next(word)
next(word)
next(word)
>>>
StopIterationTraceback (most recent call last)
<ipython-input-15-a62e3e37f15b> in <module>
2 next(word)
3 next(word)
----> 4 next(word)
StopIteration:
next(word, "default")
>>> 'default'
제너레이터 사용하기
def sequence(start=0):
while True:
yield start
start += 1
seq = sequence(10)
next(seq)
>>> 10
next(seq)
>>> 11
itertools
from itertools import islice
purchases = islice(filter(lambda p: p > 1000.0, purchases), 10)
이터레이터를 사용한 코드 간소화
def process_purchases(purchases):
# 원래의 이터러블을 세 개의 새로운 이터러블로 분신술
min_, max_, avg = itertools.tee(purchases, 3)
return min(min_), max(max_), median(avg)
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(f"{desired_value} not found")
print("%i, %i에서 값 %r 찾음" % (*coord, desired_value))
return coord
__iter__
와 __next__
매직 메서드를 구현한 객체이다.__iter__
구현 -> 이터러블__next__
구현 -> 이터레이터이터레이션 인터페이스
__iter__
를 통해 이터레이터를 반환하고, __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
# 시퀀스에서 하나씩 값을 가져올 수는 있지만 반복할 수 없다.
for _ in SequenceIterator(): pass
>>>
TypeErrorTraceback (most recent call last)
<ipython-input-32-3e62347f344e> in <module>
1 # 시퀀스에서 하나씩 값을 가져올 수는 있지만 반복할 수 없다.
----> 2 for _ in SequenceIterator(): pass
TypeError: 'SequenceIterator' object is not iterable
이터러블이 가능한 시퀀스 객체
__iter__
를 구현한 객체는 for 루프에서 사용할 수 있지만, 반복을 위해 이것을 구현해야 하는 것은 아니다.__getitem__
과 __len__
을 구현한 경우에도 반복 가능하다.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)
print(f"index {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 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__
와 __next__
를 구현한 제너레이터 객체를 이용해 단순히 다음 요소로 반복, 이동할수도 있지만 이런 기본 기능 외에도 제너레이터를 코루틴으로 활용할수도 있다..close()
.throw(ex_type[, ex_value[, ex_traceback]])
.send(value)
close()
# 코루틴 예시
def stream_db_records(db_handler):
try:
while True:
yield db_handler.read_n_records(10)
except GeneratorExit:
db_handler.close()
throw(ex_type[, ex_value[, ex_traceback]])
class CustomException(Exception):
pass
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.throw(CustomException)
이 발생한 경우 제너레이터는 INFO 레벨의 메시지를 기록하고 다음 yield 구문으로 이동한다.except Exception as e
구문이 없다면 streamer.throw(RuntimeError)
이 발생한 경우 제너레이터가 중지된 행에서 예외가 호출자에게 전파되고, 제너레이터는 중지된다.send(value)
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()
메서드를 사용하면 할당 구문 오른쪽에 yield 키워드가 나오게 되고, 할당 구문의 변수를 사용해 인자 값을 사용할 수 있다.next()
를 적어도 한 번은 호출해야 한다.# 더 간단한 버전
def stream_db_records(db_handler):
retrieved_data = None
previous_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()
를 반드시 호출해야 한다는 것을 기억할 필요 없이 코루틴을 생성하자마자 바로 사용할 수 있다면 훨씬 편할 것이다. 이를 위해 데코레이터를 사용해 코루틴을 초기화할 수 있다.# 데코레이터 사용 예시
@prepare_coroutine
def stream_db_records(db_handler):
retrieved_data = None
previous_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()
코루틴에서 값 반환하기
def generator():
yield 1
yield 2
return 3
value = generator()
next(value)
>>> 1
next(value)
>>> 2
try:
next(value)
except StopIteration as e:
print(f"returned value: {e.value}")
>>> returned value: 3
작은 코루틴에 위임하기 - 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']
서브 제너레이터에서 반환된 값 구하기
def sequence(name, start, end):
print("%s 제너레이터 %i에서 시작" % (name, start))
yield from range(start, end)
print("%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)
>>>
first 제너레이터 0에서 시작
0
next(g)
1
next(g)
>>>
first 제너레이터 5에서 종료
second 제너레이터 5에서 시작
5
next(g)
>>>
second 제너레이터 10에서 종료
StopIterationTraceback (most recent call last)
<ipython-input-77-e734f8aca5ac> in <module>
----> 1 next(g)
StopIteration: 15
a = yield b
: send()
를 통해 받은 값을 a에 할당하고 b를 호출자에게 보낸 후 멈춘다.a = yield from b
: b 이터러블에서 값을 하나씩 받아와 yield하고 b가 최종적으로 반환한 값을 a에 할당한다.서브 제너레이터와 데이터 송수신하기
def sequence(name, start, end):
value = start
print("%s 제너레이터 %i에서 시작" % (name, value))
while value < end:
try:
received = yield value
print("%s 제너레이터 %r값 수신" % (name, received))
value += 1
except CustomException as e:
print("%s 제너레이터 %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
g = main()
next(g)
>>>
first 제너레이터 0에서 시작
0
g.send("first generator")
>>>
first 제너레이터 'first generator'값 수신
1
g.throw(CustomException("custom exception"))
>>>
first 제너레이터 custom exception 에러 처리
'OK'
# 새로운 코루틴 작성 방법
import asyncio
import time
async def say_after(delay, what):
await asyncio.sleep(delay)
print(what)
async def main():
print(f"started at {time.strftime('%X')}")
await say_after(1, 'hello')
await say_after(2, 'world')
print(f"finished at {time.strftime('%X')}")
# for plain python
# asyncio.run(main())
# for jupyter - 이미 주피터 노트북이 이벤트 루프를 수행중이므로 만들어주지 않아도 됨
await main()
started at 13:14:31
hello
world
finished at 13:14:34