본문으로 건너뛰기
Advertisement

제너레이터 함수와 표현식 — yield, send(), throw()

**제너레이터(generator)**는 yield 키워드를 사용해 값을 하나씩 생성하는 특별한 함수입니다. 이터레이터를 간단하게 만들고, 메모리를 절약하며, 코루틴의 기반이 됩니다.

yield 키워드: 제너레이터 함수​

함수 본문에 yield가 있으면 그 함수는 제너레이터 함수입니다. 호출하면 제너레이터 객체를 반환합니다.

def count_up(start: int, stop: int):
    """일반 함수처럼 보이지만 yield가 있어 제너레이터 함수"""
    current = start
    while current <= stop:
        yield current   # 값 반환 + 일시 정지
        current += 1    # next() 호출 시 여기서 재개

# 호출하면 제너레이터 객체 반환 (코드 실행 안 됨)
gen = count_up(1, 5)
print(type(gen))   # <class 'generator'>
print(gen)         # <generator object count_up at 0x...>

# next()로 값 꺼내기
print(next(gen))   # 1
print(next(gen))   # 2

# for 루프로 소비
for n in count_up(1, 5):
    print(n)   # 1, 2, 3, 4, 5

일반 함수 vs 제너레이터 함수​

# 일반 함수: 리스트를 통째로 메모리에 로드
def squares_list(n: int) -> list[int]:
    return [x ** 2 for x in range(n)]

# 제너레이터: 하나씩 생성 — 메모리 O(1)
def squares_gen(n: int):
    for x in range(n):
        yield x ** 2

import sys
print(sys.getsizeof(squares_list(1_000_000)))  # ~8MB
gen = squares_gen(1_000_000)
print(sys.getsizeof(gen))                       # ~112 bytes

yield로 함수 상태 유지​

def stateful_counter():
    """yield는 함수의 상태(로컬 변수, 실행 위치)를 보존합니다"""
    count = 0
    while True:
        received = yield count   # 값 반환하고 일시 정지
        if received == "reset":
            count = 0
        else:
            count += 1

counter = stateful_counter()
print(next(counter))    # 0 (초기화)
print(next(counter))    # 1
print(next(counter))    # 2
counter.send("reset")   # 리셋
print(next(counter))    # 1

제너레이터 표현식​

리스트 컴프리헨션과 문법이 같지만 [] 대신 ()를 사용합니다.

# 리스트 컴프리헨션: 즉시 평가, 메모리에 저장
squares_list = [x**2 for x in range(10)]

# 제너레이터 표현식: 지연 평가, 메모리 절약
squares_gen = (x**2 for x in range(10))

print(type(squares_list))  # <class 'list'>
print(type(squares_gen))   # <class 'generator'>

# 소비
print(sum(x**2 for x in range(10)))         # 285 — 괄호 생략 가능
print(list(x**2 for x in range(10)))        # [0, 1, 4, 9, ...]
print(max(len(word) for word in ["hi", "hello", "hey"]))  # 5
# 조건 포함 제너레이터 표현식
evens = (x for x in range(20) if x % 2 == 0)
print(list(evens))  # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

# 중첩 제너레이터
matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
flat = (elem for row in matrix for elem in row)
print(list(flat))  # [1, 2, 3, 4, 5, 6, 7, 8, 9]

send(): 제너레이터에 값 전달​

send(value)는 next()처럼 제너레이터를 재개하면서, yield 표현식의 결과값으로 전달합니다.

def accumulator():
    """외부에서 값을 받아 누적합산"""
    total = 0
    while True:
        value = yield total   # total을 반환하고, 외부에서 받은 값을 value에 저장
        if value is None:
            break
        total += value


gen = accumulator()
next(gen)          # 제너레이터 시작 (첫 yield까지 실행), 0 반환
print(gen.send(10))   # 10 전달 → total=10 → 10 반환
print(gen.send(20))   # 20 전달 → total=30 → 30 반환
print(gen.send(5))    # 5 전달 → total=35 → 35 반환
gen.send(None)        # 종료 신호

주의: 첫 send() 호출 전에 반드시 next() 또는 send(None)으로 제너레이터를 시작해야 합니다.

# priming 데코레이터로 자동 시작
from functools import wraps
from typing import Callable, Any

def prime(gen_func: Callable) -> Callable:
    """제너레이터 함수를 자동으로 시작하는 데코레이터"""
    @wraps(gen_func)
    def wrapper(*args, **kwargs):
        gen = gen_func(*args, **kwargs)
        next(gen)  # 첫 yield까지 자동 실행
        return gen
    return wrapper


@prime
def running_average():
    """실시간 평균 계산기"""
    total, count = 0, 0
    while True:
        value = yield total / count if count > 0 else 0
        total += value
        count += 1


avg = running_average()   # 자동으로 primed
print(avg.send(10))  # 10.0
print(avg.send(20))  # 15.0
print(avg.send(30))  # 20.0

throw(): 예외 주입​

외부에서 제너레이터 내부에 예외를 던집니다.

def generator_with_error_handling():
    """외부에서 주입된 예외를 처리하는 제너레이터"""
    value = 0
    while True:
        try:
            value = yield value
        except ValueError as e:
            print(f"ValueError 잡힘: {e}")
            value = 0  # 리셋
        except GeneratorExit:
            print("제너레이터 종료 요청")
            return


gen = generator_with_error_handling()
next(gen)                          # 시작
print(gen.send(10))                # 10
print(gen.send(20))                # 20
gen.throw(ValueError, "잘못된 값")   # ValueError 주입 → 0으로 리셋
print(next(gen))                   # 1 (0+1)
gen.close()                        # GeneratorExit 발생

close(): 제너레이터 종료​

close()는 제너레이터에 GeneratorExit 예외를 던져 정상 종료시킵니다.

def resource_holding_gen():
    """리소스를 보유하는 제너레이터"""
    print("리소스 획득")
    try:
        while True:
            yield "데이터"
    finally:
        print("리소스 해제")  # close() 호출 시 반드시 실행됨


gen = resource_holding_gen()
print(next(gen))   # "데이터"
gen.close()        # GeneratorExit → finally 실행 → "리소스 해제"

무한 시퀀스, 지연 평가 활용​

def naturals(start: int = 1):
    """무한 자연수 시퀀스"""
    n = start
    while True:
        yield n
        n += 1

def primes():
    """에라토스테네스의 체 — 무한 소수 시퀀스"""
    composites: set[int] = set()
    for n in naturals(2):
        if n not in composites:
            yield n
            composites.update(range(n * n, n * n + n * 100, n))

import itertools

# 처음 10개 소수
first_10_primes = list(itertools.islice(primes(), 10))
print(first_10_primes)  # [2, 3, 5, 7, 11, 13, 17, 19, 23, 29]

# 100 미만 소수
small_primes = list(itertools.takewhile(lambda x: x < 100, primes()))
print(small_primes)

실전 예제: 데이터 파이프라인​

from pathlib import Path
import csv
from typing import Iterator

def read_csv_rows(filepath: str | Path) -> Iterator[dict]:
    """CSV 파일 행을 하나씩 생성"""
    with open(filepath, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield dict(row)

def filter_active(rows: Iterator[dict]) -> Iterator[dict]:
    """활성 사용자만 필터링"""
    for row in rows:
        if row.get("status") == "active":
            yield row

def enrich_data(rows: Iterator[dict]) -> Iterator[dict]:
    """데이터 변환/보강"""
    for row in rows:
        row["full_name"] = f"{row['first_name']} {row['last_name']}"
        row["age"] = int(row.get("age", 0))
        yield row

def batch_rows(rows: Iterator[dict], size: int = 100) -> Iterator[list[dict]]:
    """배치 단위로 묶기"""
    batch = []
    for row in rows:
        batch.append(row)
        if len(batch) >= size:
            yield batch
            batch = []
    if batch:
        yield batch

def process_users(csv_file: str) -> None:
    """파이프라인 조합"""
    pipeline = batch_rows(
        enrich_data(
            filter_active(
                read_csv_rows(csv_file)
            )
        ),
        size=50
    )

    for batch_num, batch in enumerate(pipeline, 1):
        print(f"배치 {batch_num}: {len(batch)}명 처리")
        for user in batch:
            save_to_db(user)

# 어떤 CSV 크기에도 메모리 O(batch_size)로 처리 가능
process_users("million_users.csv")

고수 팁​

팁 1: yield from으로 서브제너레이터 위임

def chain(*iterables):
    for it in iterables:
        yield from it   # 각 이터러블의 모든 값을 위임

print(list(chain([1, 2], [3, 4], [5])))  # [1, 2, 3, 4, 5]

팁 2: 제너레이터 반환값 (return + StopIteration.value)

def gen_with_return():
    yield 1
    yield 2
    return "완료"   # StopIteration의 value로 전달

g = gen_with_return()
next(g)  # 1
next(g)  # 2
try:
    next(g)
except StopIteration as e:
    print(e.value)  # "완료"

팁 3: inspect.isgeneratorfunction()으로 제너레이터 함수 확인

import inspect

def normal_func(): return 42
def gen_func(): yield 42

print(inspect.isgeneratorfunction(normal_func))  # False
print(inspect.isgeneratorfunction(gen_func))     # True
Advertisement
Advertisement