본문으로 건너뛰기
Advertisement

yield from과 위임 제너레이터

yield from은 Python 3.3에서 PEP 380으로 도입된 문법으로, 제너레이터가 다른 이터러블이나 서브제너레이터에게 작업을 위임합니다. 코루틴의 기반이 되는 강력한 기능입니다.

yield from 기본 동작​

yield from iterable은 이터러블의 모든 값을 현재 제너레이터를 통해 그대로 전달합니다.

# yield from 없이 서브이터러블 값 전달
def chain_without_yield_from(*iterables):
    for it in iterables:
        for value in it:   # 중첩 반복
            yield value

# yield from으로 간결하게
def chain_with_yield_from(*iterables):
    for it in iterables:
        yield from it      # 위임!

result = list(chain_with_yield_from([1, 2], [3, 4], [5, 6]))
print(result)  # [1, 2, 3, 4, 5, 6]
# 문자열, 리스트 등 모든 이터러블에 적용 가능
def flatten(nested):
    for item in nested:
        if isinstance(item, (list, tuple)):
            yield from flatten(item)  # 재귀적 위임
        else:
            yield item

data = [1, [2, 3], [4, [5, 6]], 7]
print(list(flatten(data)))  # [1, 2, 3, 4, 5, 6, 7]

값 전달과 예외 전파​

yield from의 진짜 강점은 단순한 값 전달을 넘어 양방향 채널을 제공한다는 점입니다.

# 위임자(delegating generator): yield from을 사용하는 제너레이터
# 서브제너레이터(subgenerator): yield from이 가리키는 제너레이터

def subgenerator():
    """서브제너레이터"""
    received = yield "first"
    print(f"서브제너레이터가 받은 값: {received}")
    received = yield "second"
    print(f"서브제너레이터가 받은 값: {received}")
    return "서브제너레이터 완료"   # return 값이 위임자에게 전달됨


def delegating_gen():
    """위임 제너레이터"""
    result = yield from subgenerator()   # 서브제너레이터의 return 값 수신
    print(f"서브제너레이터 결과: {result}")
    yield "after delegation"


gen = delegating_gen()
print(next(gen))       # "first"
print(gen.send("A"))   # 서브제너레이터가 받은 값: A → "second"
print(gen.send("B"))   # 서브제너레이터가 받은 값: B → 서브제너레이터 완료 → "after delegation"

yield from이 처리하는 것:

  • 서브제너레이터로 next() 전달
  • send(value) 값을 서브제너레이터에 전달
  • throw(exc) 예외를 서브제너레이터에 전달
  • 서브제너레이터의 StopIteration.value를 yield from 표현식의 결과로 반환

재귀적 제너레이터 패턴​

from typing import Any, Iterator

# 중첩 구조 평탄화
def deep_flatten(nested: Any) -> Iterator:
    """임의 깊이의 중첩 구조를 평탄화"""
    if isinstance(nested, (list, tuple, set)):
        for item in nested:
            yield from deep_flatten(item)
    else:
        yield nested

data = [1, [2, [3, [4, [5]]]]]
print(list(deep_flatten(data)))  # [1, 2, 3, 4, 5]


# 트리 구조 순회
from dataclasses import dataclass, field

@dataclass
class TreeNode:
    value: int
    children: list["TreeNode"] = field(default_factory=list)


def preorder(node: TreeNode) -> Iterator[int]:
    """전위 순회 (루트 → 왼쪽 → 오른쪽)"""
    yield node.value
    for child in node.children:
        yield from preorder(child)   # 재귀적 위임


def postorder(node: TreeNode) -> Iterator[int]:
    """후위 순회 (왼쪽 → 오른쪽 → 루트)"""
    for child in node.children:
        yield from postorder(child)
    yield node.value


# 트리 생성
root = TreeNode(1, [
    TreeNode(2, [TreeNode(4), TreeNode(5)]),
    TreeNode(3, [TreeNode(6)])
])

print(list(preorder(root)))   # [1, 2, 4, 5, 3, 6]
print(list(postorder(root)))  # [4, 5, 2, 6, 3, 1]

코루틴과의 관계​

yield from은 asyncio 이전에 코루틴을 구현하는 기반 메커니즘이었습니다. Python 3.5+의 async/await는 yield from 위에 구축됩니다.

# Python 3.4 스타일 코루틴 (이해를 위해 — 실제로는 async/await 사용)
import asyncio

def old_style_coroutine():
    """yield from 기반 코루틴 (역사적 이해용)"""
    result = yield from asyncio.sleep(1)  # asyncio와 통신
    return result

# 현대적 스타일 (실제 사용)
async def modern_coroutine():
    result = await asyncio.sleep(1)  # async/await = syntactic sugar
    return result
# yield from의 예외 전파 확인
def sub_gen():
    try:
        yield 1
        yield 2
    except RuntimeError as e:
        print(f"서브에서 잡힘: {e}")
        yield "오류 후 복구"


def delegator():
    yield from sub_gen()


gen = delegator()
print(next(gen))                           # 1
gen.throw(RuntimeError, "외부 오류 주입")   # 서브에서 잡힘: 외부 오류 주입
# "오류 후 복구" 출력 후 StopIteration

실전: 트리 순회, 파이프라인​

파일 시스템 순회​

from pathlib import Path
from typing import Iterator


def iter_files(directory: str | Path, pattern: str = "*") -> Iterator[Path]:
    """디렉터리 내 모든 파일 재귀 순회"""
    root = Path(directory)
    for entry in root.iterdir():
        if entry.is_dir():
            yield from iter_files(entry, pattern)   # 하위 디렉터리 위임
        elif entry.match(pattern):
            yield entry


# Python 파일 찾기
for py_file in iter_files("project/", "*.py"):
    print(py_file)

제너레이터 파이프라인 조합​

from typing import Iterator, Callable, TypeVar

T = TypeVar("T")


def pipeline(*generators: Callable) -> Callable:
    """여러 제너레이터를 파이프라인으로 연결하는 유틸리티"""
    def apply(source):
        result = source
        for gen in generators:
            result = gen(result)
        return result
    return apply


def read_lines(filepath: str) -> Iterator[str]:
    with open(filepath, encoding="utf-8") as f:
        yield from f


def strip_lines(lines: Iterator[str]) -> Iterator[str]:
    for line in lines:
        stripped = line.strip()
        if stripped:
            yield stripped


def parse_csv_line(lines: Iterator[str]) -> Iterator[list[str]]:
    import csv
    reader = csv.reader(lines)
    yield from reader


def filter_header(rows: Iterator[list[str]]) -> Iterator[list[str]]:
    first = True
    for row in rows:
        if first:
            first = False
            continue
        yield row


# 파이프라인 조합
process = pipeline(strip_lines, parse_csv_line, filter_header)
data = process(read_lines("data.csv"))

for row in data:
    print(row)

재귀적 JSON 순회​

from typing import Any, Iterator


def iter_values(data: Any, path: str = "") -> Iterator[tuple[str, Any]]:
    """중첩 JSON 구조의 모든 리프 노드를 (경로, 값) 형태로 순회"""
    if isinstance(data, dict):
        for key, value in data.items():
            new_path = f"{path}.{key}" if path else key
            yield from iter_values(value, new_path)
    elif isinstance(data, list):
        for i, item in enumerate(data):
            yield from iter_values(item, f"{path}[{i}]")
    else:
        yield path, data


config = {
    "database": {
        "host": "localhost",
        "port": 5432,
        "credentials": {
            "user": "admin",
            "password": "secret"
        }
    },
    "features": ["auth", "billing", "analytics"]
}

for path, value in iter_values(config):
    print(f"{path}: {value}")
# database.host: localhost
# database.port: 5432
# database.credentials.user: admin
# database.credentials.password: secret
# features[0]: auth
# features[1]: billing
# features[2]: analytics

고수 팁​

팁 1: yield from range(n)은 for x in range(n): yield x보다 빠름

# 느린 방법
def slow_gen(n):
    for x in range(n):
        yield x

# 빠른 방법
def fast_gen(n):
    yield from range(n)

팁 2: 서브제너레이터의 return 값 활용

def producer():
    yield 1
    yield 2
    return {"count": 2, "status": "완료"}

def consumer():
    stats = yield from producer()
    print(f"통계: {stats}")
    yield "done"

list(consumer())
# 통계: {'count': 2, 'status': '완료'}

팁 3: yield from으로 제너레이터 체인 구성

import itertools

def merged(*generators):
    """여러 제너레이터를 순서대로 연결"""
    for gen in generators:
        yield from gen

# itertools.chain과 동일하지만 커스텀 로직 추가 가능
result = list(merged(range(3), range(5, 8)))
print(result)  # [0, 1, 2, 5, 6, 7]
Advertisement
Advertisement