Pythonジェネレータの基本と使い方: 遅延評価で省メモリ処理

# 最小のジェネレータ例
def simple_gen():
    # 値を1つずつ「遅延生成」して返す
    yield 1
    yield 2
    yield 3

g = simple_gen()
print(next(g))  # 1つ目を取り出す
print(next(g))  # 2つ目
print(next(g))  # 3つ目

# これ以上は要素がないので StopIteration が送出される
try:
    print(next(g))
except StopIteration:
    print("StopIteration が発生しました")

# リスト内包表記は全要素をメモリに載せる
squares_list = [x * x for x in range(5)]  # [0, 1, 4, 9, 16]

# ジェネレータ式は要素を都度生成
squares_gen = (x * x for x in range(5))

print("list:", squares_list)
print("gen 1つ目:", next(squares_gen))
print("gen 2つ目:", next(squares_gen))

def up_to(n):
    for i in range(n):
        yield i

g = up_to(3)

# next()で手動消費
print(next(g))  # 0
print(next(g))  # 1

# 残りをforで消費
for x in g:
    print("for:", x)  # 2 が出力されて終了

def nums():
    for i in range(3):
        yield i

g = nums()
print(list(g))  # [0, 1, 2] で使い切り
print(list(g))  # 空になる(すでに消費済み)

from io import StringIO

def iter_lines(handle):
    # with のスコープはジェネレータの寿命に合わせて維持されます
    # 実ファイルなら: with open(path, "r", encoding="utf-8") as f:
    with handle as f:
        for line in f:
            line = line.rstrip("\n")
            # 空行をスキップしつつ、必要なら加工
            if line:
                yield line

# デモ用: メモリ上のファイル風オブジェクト
fake_file = StringIO("alpha\n\nbeta\ngamma\n")

for ln in iter_lines(fake_file):
    print(ln)

from io import StringIO

def read_lines(handle):
    with handle as f:
        for line in f:
            yield line.strip()

def parse_ints(lines):
    for s in lines:
        if s:  # 空行を除外
            # 変換に失敗したものはスキップ
            try:
                yield int(s)
            except ValueError:
                continue

def filter_positive(nums):
    for n in nums:
        if n > 0:
            yield n

def square(nums):
    for n in nums:
        yield n * n

# デモ入力
fake = StringIO("10\n0\n-3\nfoo\n5\n")

pipeline = square(filter_positive(parse_ints(read_lines(fake))))
for v in pipeline:
    print(v)

def chunked(iterable, size):
    """iterableをsize件ずつのリストにまとめて逐次返すジェネレータ"""
    chunk = []
    for item in iterable:
        chunk.append(item)
        if len(chunk) >= size:
            # ここでチャンクを1つ返してリセット
            yield chunk
            chunk = []
    if chunk:
        # 端数があれば最後に返す
        yield chunk

# 例: 100件のIDを10件ずつDBにバルク挿入するイメージ
ids = range(1, 21)  # デモとして20件
for group in chunked(ids, size=6):
    print("insert bulk:", group)

def fib():
    """フィボナッチの無限列"""
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# 先頭10個だけ消費
from itertools import islice
first10 = list(islice(fib(), 10))
print(first10)

def numbers(n):
    for i in range(n):
        yield i * i

# 悪い例: 最初に全部リスト化してしまう
# big = list(numbers(1_000_000))

# 良い例: ストリーミングで合計だけ計算
total = sum(numbers(1_000_000))
print("sum:", total)

from itertools import tee

def gen():
    for i in range(5):
        yield i

g1, g2 = tee(gen(), 2)
print(list(g1))  # [0, 1, 2, 3, 4]
print(list(g2))  # [0, 1, 2, 3, 4]

def managed_resource():
    print("open resource")
    try:
        for i in range(3):
            yield i
    finally:
        # 例外やclose()呼び出し時に必ず実行される
        print("cleanup resource")

g = managed_resource()
print(next(g))      # open
g.close()           # 明示クローズ -> finally 実行

# 再デモ: 途中で例外が発生するケース
g2 = managed_resource()
try:
    for x in g2:
        print("use:", x)
        if x == 1:
            raise RuntimeError("oops")
except RuntimeError:
    print("handled error")

import time
import tracemalloc

N = 2_000_00  # 20万 (デモ用に控えめ)

def calc_list():
    # 全要素を先に作る
    return sum([i * i for i in range(N)])

def calc_gen():
    # 都度生成して合計
    return sum(i * i for i in range(N))

def timed(func):
    tracemalloc.start()
    t0 = time.perf_counter()
    result = func()
    t1 = time.perf_counter()
    current, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return {
        "result": result,
        "time_sec": t1 - t0,
        "peak_kb": peak / 1024.0,
    }

print("list:", timed(calc_list))
print("gen :", timed(calc_gen))