generator 짤 줄 알고 itertools 쓸 줄 알면 — 파이프라인 만들 수 있어. 각 단계가 generator, iterable 받아서 변환된 iterable yield. Unix pipe 처럼 합성. 전체 파이프라인이 lazy — consumer 가 값 당기기 전엔 아무 일도 안 일어나, 그 다음엔 정확히 한 값이 모든 단계 통과.
nested 루프를 이기는 모양
좋은 Python 에서 자주 봐 — 다섯 가지 일 하는 큰 루프 하나 대신, 각각 한 가지 일 하는 작은 generator 함수 다섯 개를 체인. parsed = parse(stripped(non_empty(lines(path)))). 각 layer 테스트 가능, 각 재사용 가능, 데이터는 한 번만 걸어.
materialize-everything 함정 피하기
실수는 중간에 list() 호출해서 체인 깨는 거. list(stage_3(stage_2(stage_1(...)))) OK — 그게 consumer. 근데 stage_3(list(stage_2(stage_1(...)))) 는 목적 패배 — 중간 결과 materialize, 안 써도 될 메모리 먹어. 진짜 list 필요할 때까지 끝까지 lazy 유지.
파이프라인과 부작용 안 어울려
generator 본문에 부작용 (로깅, 공유 state 변경, 예외 raise) 있으면 — 그 부작용은 *반복 시점* 에 일어나, 정의 시점 X. g = my_generator() 가 아무것도 안 돌려. 첫 next(g) 가 본문을 첫 yield 까지 돌려. 알면 OK, 모르면 디버깅 악몽.
War Story: 흔한 버그 — generator 의 __anext__ 호출을 asyncio.wait_for 로 감싸기. 타임아웃 나면 cancellation 이 generator 종료 — 이후 next 가 복구 대신 StopIteration. cwkPippa 의 어댑터는 일부러 안 그래. 교훈 — 반복에 대한 타임아웃은 미묘함, 가벼운 데코레이터가 아니라 load-bearing 구조 코드처럼 다뤄.
Code
파이프라인 — 5 단계 stream·python
def lines_of(text):
yield from text.splitlines()
def strip_each(lines):
for line in lines:
yield line.strip()
def drop_empty(lines):
for line in lines:
if line:
yield line
def first_word(lines):
for line in lines:
yield line.split(" ", 1)[0]
def long_only(words, min_len=4):
for w in words:
if len(w) >= min_len:
yield w
source = """hello world\n\n foo bar\nhi there\nlongword something"""
result = list(long_only(first_word(drop_empty(strip_each(lines_of(source))))))
print(result) # ['hello', 'longword']
Lazy 가 eager 이김 — source 가 거대할 때·python
import itertools as it
def naturals():
n = 1
while True:
yield n
n += 1
def squares(nums):
for n in nums:
yield n * n
def under_limit(nums, limit):
for n in nums:
if n >= limit:
return
yield n
# 파이프라인: naturals -> squares -> 10000 미만
result = list(under_limit(squares(naturals()), 10000))
print(result)
# [1, 4, 9, 16, 25, 36, 49, 64, 81, 100, ...]
# 99*99 = 9801 에서 멈춤 (다음은 10000)
버그 — generator 는 정의 시점에 안 돌아·python
def risky():
print("about to fail")
raise ValueError("boom")
yield 1 # 도달 X, 근데 함수는 여전히 generator
g = risky() # 출력 X, 에러 X — 본문 아직 안 돌아
print("got generator")
try:
next(g)
except ValueError as e:
print("raised at iteration time:", e)
# about to fail
# raised at iteration time: boom
중간 materialize 가 lazy 패배·python
import itertools as it
def big_source():
for i in range(10**6):
yield i
# 안 좋음 — 슬라이싱 전에 백만 원소 list 통째 materialize
result_bad = list(it.islice(list(big_source()), 5))
# 좋음 — 끝까지 lazy
result_good = list(it.islice(big_source(), 5))
print(result_bad == result_good) # True — 같은 답
# 근데 result_bad 가 가는 길에 백만 원소 list 할당
멀티라인 CSV 같은 문자열 'name,age\nalice,30\nbob,25\ncharlie,99\n,' 위에 4 단계 generator 파이프라인. 단계 1 — 비-빈 줄 strip. 단계 2 — 콤마 적어도 하나 있는 줄만. 단계 3 — (name, int(age)) 튜플, 변환 실패한 줄은 skip. 단계 4 — age >= 18 인 튜플만. list 로 소비. 입력에서 작동하고 실패하는 줄이 조용히 skip 되는지 확인.
Progress
Progress is local-only — sign in to sync across devices.