Pythonのジェネレータ入門：yieldの挙動・メリット・使い所を図解で理解

def simple_generator():
    print("start")
    yield 1          # 1回目の値
    print("between")
    yield 2          # 2回目の値
    print("end")

gen = simple_generator()   # まだ何も実行されていない

print(next(gen))           # ここで「start」が表示され、1が返る
print(next(gen))           # ここで「between」が表示され、2が返る

try:
    print(next(gen))       # ここで「end」が表示され、その後StopIterationが送出される
except StopIteration:
    print("finished")

def counter(n):
    print("generator started")
    for i in range(n):
        print(f"yielding {i}")
        yield i
    print("generator finished")

for value in counter(3):
    print("got:", value)

def gen_with_return():
    yield 1
    yield 2
    return 99   # ここでStopIteration(value=99)が送出される

g = gen_with_return()
print(next(g))      # 1
print(next(g))      # 2
try:
    print(next(g))  # ここでStopIteration例外
except StopIteration as e:
    print("stopped with value:", e.value)

# リスト内包表記(すべての値をリストとして生成)
lst = [x * 2 for x in range(5)]
print(lst)

# ジェネレータ式(値を1つずつ生成するジェネレータを返す)
gen = (x * 2 for x in range(5))
print(gen)          # ジェネレータオブジェクトの表示
print(list(gen))    # ここで初めて値が生成される

# リストで全件をメモリに載せる場合
def square_list(n):
    result = []
    for i in range(n):
        result.append(i * i)
    return result

# ジェネレータで1件ずつ返す場合
def square_generator(n):
    for i in range(n):
        yield i * i

N = 10**6

# リスト版
lst = square_list(N)        # ここで100万件分のリストがメモリに載る
print("list length:", len(lst))

# ジェネレータ版
gen = square_generator(N)
count = 0
for _ in gen:
    count += 1
print("generated count:", count)

def read_numbers(lines):
    """文字列行から整数を読み取るジェネレータ"""
    for line in lines:
        line = line.strip()
        if not line:
            continue
        yield int(line)

def filter_even(numbers):
    """偶数だけを通すジェネレータ"""
    for n in numbers:
        if n % 2 == 0:
            yield n

def square(numbers):
    """2乗に変換するジェネレータ"""
    for n in numbers:
        yield n * n

# パイプラインとしてつなげる
def process(lines):
    nums = read_numbers(lines)
    evens = filter_even(nums)
    squared = square(evens)
    return squared

data = ["1", "2", "3", "4", "", "5"]
for value in process(data):
    print(value)

def naturals():
    n = 0
    while True:   # 無限ループ
        yield n
        n += 1

# 最初の5個だけ使う
gen = naturals()
for _ in range(5):
    print(next(gen))

def read_large_file(path):
    """ファイルを1行ずつ読み込むジェネレータ"""
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            yield line.rstrip("\n")

def search_keyword(lines, keyword):
    """特定のキーワードを含む行だけを返すジェネレータ"""
    for line in lines:
        if keyword in line:
            yield line

def example():
    logfile = "access.log"
    lines = read_large_file(logfile)
    error_lines = search_keyword(lines, "ERROR")
    for line in error_lines:
        print(line)

# example() を呼ぶと、巨大ファイルでも少ないメモリで検索できる

import requests

def fetch_items(api_url):
    """ページネーションされたAPIから、アイテムを逐次取得するジェネレータ"""
    page = 1
    while True:
        params = {"page": page}
        resp = requests.get(api_url, params=params)
        resp.raise_for_status()
        data = resp.json()
        items = data["items"]
        if not items:
            # もうデータがなければ終了
            return
        for item in items:
            yield item
        page += 1

# 利用例
# for item in fetch_items("https://api.example.com/items"):
#     print(item)

def normalize(records):
    """レコードのフィールド名を小文字に揃える"""
    for rec in records:
        yield {k.lower(): v for k, v in rec.items()}

def filter_active(records):
    """status が active のものだけに絞り込む"""
    for rec in records:
        if rec.get("status") == "active":
            yield rec

def project_fields(records, fields):
    """必要なフィールドだけを残す"""
    for rec in records:
        yield {k: rec.get(k) for k in fields}

def pipeline(records):
    step1 = normalize(records)
    step2 = filter_active(step1)
    step3 = project_fields(step2, ["id", "name"])
    return step3

def example():
    raw = [
        {"ID": 1, "Name": "Alice", "Status": "active"},
        {"ID": 2, "Name": "Bob",   "Status": "inactive"},
        {"ID": 3, "Name": "Carol", "Status": "active"},
    ]
    for rec in pipeline(raw):
        print(rec)

# example() を呼び出した時の出力イメージ:
# {'id': 1, 'name': 'Alice'}
# {'id': 3, 'name': 'Carol'}

class Node:
    def __init__(self, value, children=None):
        self.value = value
        self.children = children or []

def dfs(node):
    """深さ優先探索でノードを順にyieldするジェネレータ"""
    yield node
    for child in node.children:
        # 再帰ジェネレータからの値をまとめてyieldする
        yield from dfs(child)

def example():
    # ツリー構造を構築
    #       A
    #      / \
    #     B   C
    #    /   / \
    #   D   E   F
    nD = Node("D")
    nB = Node("B", [nD])
    nE = Node("E")
    nF = Node("F")
    nC = Node("C", [nE, nF])
    nA = Node("A", [nB, nC])

    for node in dfs(nA):
        print(node.value)

# example() の出力:
# A
# B
# D
# C
# E
# F

def collect_errors(lines):
    errors = []
    for line in lines:
        if "ERROR" in line:
            errors.append(line)
    return errors
これをジェネレータ化すると次のようになります。def iter_errors(lines):
    for line in lines:
        if "ERROR" in line:
            yield line
呼び出し側がリストを期待しているなら、当面はlist()で包んで互換性を保てます。# 既存インターフェースを一時的に維持
def collect_errors(lines):
    return list(iter_errors(lines))