Python文法講義~ユーザー定義関数応用~

まえがき

今回からPythonの文法事項について解説していく講義形式の記事を書く。どちらかというと、初心者向けテキストでは扱われることが少ないテーマを中心に扱うことを目的としているが、一方でこの記事だけでも、その理解に前提となる内容も含めて理解が完結するように工夫も行う。以前にデコレーターを目指して記事を書いたが、個人的にはもっと書くべきこともあったかと思うので、再度応用編という形でまとめるとともに、他には扱わなかったテーマもここで回収する。

高階関数

デコレーターに話を持っていくならば、高階関数から話を始めるべきである。ただし、過去の記事でも扱っているので、ダイジェスト気味になる。

定義

関数を引数に取る、または関数を戻り値として返す関数のこと。
Pythonは「関数が第一級オブジェクト（first-class）」であるから、変数に入れる・引数で渡す・返す・データ構造に格納といった操作が可能。これが高階関数の土台になる。

高階関数の利用によって、以下のような利点がある
・抽象化：処理の枠組みと可変部（関数）を分離できるため、設計が明確化する。
・再利用性：条件や変換のみ差し替えることで同一フレームワークを流用できる。
・可読性：意図（変換・選別・集約）が関数シグネチャに明示化される。
・合成容易性：小関数を合成し大きな処理を段階的に構築できる。

主要API(標準ライブラリ)

map(func, iterable)：要素変換

nums = [1, 2, 3, 4]
def square(x): 
    return x * x

list(map(square, nums))
# 実行結果: [1, 4, 9, 16]

単射変換を宣言的に表現できる。内包表記 [(x*x) for x in nums] でも代替可能。

filter(func, iterable)：要素選別

def is_even(x): 
    return x % 2 == 0

list(filter(is_even, range(7)))
# 実行結果: [0, 2, 4, 6]

述語関数の差し替えで条件を容易に切り替えられる。

functools.reduce(func, iterable[, initial])：畳み込み

from functools import reduce

nums = [1, 2, 3, 4]
def add(a, b): 
    return a + b

reduce(add, nums)          # (((1+2)+3)+4)
# 実行結果: 10

reduce(add, nums, 100)     # 初期値を与える
# 実行結果: 110

総和・総積・逐次更新・マージ処理などに適する。

sorted(..., key=func) / min / max の key

words = ["aa", "b", "cccc", "ddd"]
sorted(words, key=len)
# 実行結果: ['b', 'aa', 'ddd', 'cccc']

max(words, key=len)
# 実行結果: 'cccc'

比較基準を関数注入する典型例である。

any と all（内包と併用）

nums = [10, 3, 6, 8]
any(n % 2 == 1 for n in nums)
# 実行結果: True

all(n < 10 for n in nums)
# 実行結果: False

functools.partial：部分適用

from functools import partial

def power(base, exp): 
    return base ** exp

square = partial(power, exp=2)
square(5)
# 実行結果: 25

引数の前固定によりパラメータ化された関数を簡潔に生成できる。

実行例

パラメータ化された述語の生成

def make_threshold_filter(th):
    def _pred(x):
        return x >= th
    return _pred  # クロージャを返す

data = [3, 7, 5, 9]
ge5 = make_threshold_filter(5)
list(filter(ge5, data))
# 実行結果: [7, 5, 9]

設定値付きの関数を動的に生成できる。

関数合成

def compose(f, g):
    def h(x):
        return f(g(x))
    return h

def inc(x): return x + 1
def double(x): return x * 2

f = compose(double, inc)  # f(x) = double(inc(x))
f(3)
# 実行結果: 8

小規模パイプラインの例

from functools import reduce
from operator import add

def make_pipeline(transformers, predicate, aggregator, initial=None):
    """
    transformers: [x -> x'] の列
    predicate   : x -> bool
    aggregator  : (acc, x) -> acc
    initial     : 初期値（Noneなら非指定）
    """
    def run(data):
        for f in transformers:
            data = map(f, data)        # 遅延変換
        data = filter(predicate, data) # 遅延選別
        return (reduce(aggregator, data, initial)
                if initial is not None else
                reduce(aggregator, data))
    return run

# 例： (x*2)→(x+1) と変換→奇数のみ選別→加算集約
pipeline = make_pipeline(
    transformers=[lambda x: x*2, lambda x: x+1],
    predicate=lambda x: x % 2 == 1,
    aggregator=add,
    initial=0,
)

pipeline(range(5))  # 0..4
# 1 + 5 + 9 + 13 + 17 = 45
# 実行結果: 45

pythonicな運用

・内包表記の活用：可読性の観点から map/filter は内包表記で置換可能な場合が多い（ただし遅延が不要であるとき）。
例：[f(x) for x in xs if pred(x)]
・key パラメータの徹底：sorted/min/max/heapq 等で比較基準を関数注入することにより、意図が明瞭になる。
・lambda は短く：式が長くなる場合は名前付き関数へ切り出し、可読性を確保すべきである。
・partial による前固定：共通パラメータを固定した関数群を迅速に構築できる。
・例外処理はデコレータへ：横断的なエラーハンドリング・ロギングはデコレータで集約すると保守性が高い。
・数値処理はベクトル化優先：性能重視の場面では NumPy/Pandas のベクトル化演算を優先し、HOFは構成・前後処理に用いるのが合理的である。

クロージャ

定義

クロージャとは、関数が定義されたときの外側スコープ中の変数（自由変数）を、その関数オブジェクトが生存する限り保持し続ける仕組みおよび、その性質を持つ関数オブジェクトそのものを指す概念である。
Pythonでは関数は第一級オブジェクトであり、関数を返す関数を定義した際に外側スコープの値を「閉じ込める（close over）」ことで状態を保持できる。

仕組み(LEGB原則と自由変数)

名前解決は L E G B 順序で行われる（Local → Enclosing → Global → Builtins）。
クロージャは Enclosing（内側関数から見た外側のローカルスコープ）にある 自由変数への参照をセル（cell）として保持する。このため、外側関数の実行が終了しても、内側関数が生きている限り、値は失われない。

最小例：関数を返す関数

def make_adder(offset: int):
    def add(x: int) -> int:
        return x + offset  # ← 外側スコープの変数 offset を参照（自由変数）
    return add

f = make_adder(10)
g = make_adder(100)

print(f(3))   # 実行結果: 13
print(g(3))   # 実行結果: 103

f と g は同じ関数定義を共有するが、それぞれ異なる offset を保持する独立のクロージャである。

nonlocal と global の役割

・nonlocal：外側のローカルスコープ（Enclosing）にある変数へ再代入したいときに用いる。
・global：モジュールグローバルへ再代入したいときに用いる。

ミニマムな例

def make_counter(start: int = 0):
    count = start
    def inc(step: int = 1) -> int:
        nonlocal count     # ← 外側ローカルを再束縛
        count += step
        return count
    return inc

c = make_counter(10)
print(c())      # 実行結果: 11
print(c(5))     # 実行結果: 16

nonlocal を付けないと、count += step は内側スコープに新規ローカル count を作ろうとして UnboundLocalError を引き起こすため注意が必要である。

テキストによく現れる典型的な例

from collections.abc import Callable

def counter(init: int) -> Callable[[], int]:
    # カウント値
    count = init

    # カウント値をインクリメントする内部関数
    def increment() -> int:
        nonlocal count
        count += 1
        return count
    return increment

c1 = counter(1)
c2 = counter(25)
print(c1())   # 2
print(c1())   # 3
print(c2())   # 26
print(c2())   # 27

典型的な用途

関数ファクトリ(パラメータ固定)

def make_threshold_pred(th: float):
    def pred(x: float) -> bool:
        return x >= th
    return pred

is_pass = make_threshold_pred(60)
print(list(filter(is_pass, [55, 60, 90])))  # 実行結果: [60, 90]

設定値付きの判定器・変換器を動的に生成できる。

状態を持つ関数(計量オブジェクト代替)

def moving_average(window: int):
    buf: list[float] = []
    def add(x: float) -> float:
        buf.append(x)
        if len(buf) > window:
            buf.pop(0)
        return sum(buf) / len(buf)
    return add

ma3 = moving_average(3)
print([ma3(x) for x in [10, 20, 30, 40]])
# 実行結果: [10.0, 15.0, 20.0, 30.0]

メモ化(簡易キャッシュ)

def memoize_last():
    last_args = None
    last_result = None
    def decorate(func):
        def wrapper(*args):
            nonlocal last_args, last_result
            if args == last_args:
                return last_result
            last_args = args
            last_result = func(*args)
            return last_result
        return wrapper
    return decorate

@memoize_last()
def expensive(x: int) -> int:
    print("compute...")
    return x * x

print(expensive(9))  # 実行結果: "compute..." 改行 81
print(expensive(9))  # 実行結果: 81  （計算スキップ）

内部の観察（__closure__ とセル）

クロージャは自由変数を cell として保持する。func.__closure__ で観察できる。

def outer(a):
    b = a * 2
    def inner(x):
        return x + a + b
    return inner

f = outer(10)
print(f.__closure__[0].cell_contents)  # 実行結果: 10  （a）
print(f.__closure__[1].cell_contents)  # 実行結果: 20  （b）
print(f(5))                            # 実行結果: 35

よくある落とし穴とその対策

ループ変数の遅延束縛(late binding)

内側関数が参照するのは最終的な変数であり、意図せず全て同じ値になることがある。

funcs = []
for i in range(3):
    funcs.append(lambda: i)  # i は後で参照される

print([f() for f in funcs])  # 実行結果: [2, 2, 2]（想定外）

回避策1：デフォルト引数で値を束縛

funcs = []
for i in range(3):
    funcs.append(lambda i=i: i)

print([f() for f in funcs])  # 実行結果: [0, 1, 2]

回避策2：小さなファクトリで閉じ込める

def make_f(v): 
    return lambda: v

funcs = [make_f(i) for i in range(3)]
print([f() for f in funcs])  # 実行結果: [0, 1, 2]

ミュータブル共有の副作用

クロージャが共有するリストや辞書を別の呼び出しでも更新してしまうことがある。必要に応じてコピー、イミュータブル化（tuple, frozenset）、もしくは関数内で再束縛（nonlocal）して制御すべきである。

デバッグ困難さ

状態が関数内部に隠れるため、過度な状態保持は可観測性を下げる。必要に応じてログやプロパティを持つクラスへの置換を検討すべきである。

実装上の指針（設計の勘所）

・「短いロジック + 明確な引数」を閉じ込めるのが適切。複雑な状態管理はクラスの方が可読であることが多い。
→ クロージャ：小さな状態と短いロジック、関数合成、関数ファクトリ、デコレータなどに適する。定義と使用が近接し、意図が明白。
→ クラス：状態が多い、ライフサイクル管理が必要、インターフェースが複数ある、継承やプロトコルに参加する、といった場合に適する。
・ nonlocal の使用は最小限にし、データフローが追いやすい形に保つべきである。
・ テスト容易性を確保するため、クロージャの外に純粋関数を分離し、クロージャは「パラメータ固定・状態保持」の薄い殻に留めるとよい。
・ 部分適用（functools.partial）で代替できるときは可読性・再利用性の面で有利なことがある。

少し実践的な例

検証器の合成（関数合成 + パラメータ注入）

def at_least(n):
    def pred(x): return x >= n
    return pred

def at_most(n):
    def pred(x): return x <= n
    return pred

def both(p, q):
    def pred(x): return p(x) and q(x)
    return pred

is_score_ok = both(at_least(60), at_most(100))
print([x for x in [55, 60, 100, 120] if is_score_ok(x)])
# 実行結果: [60, 100]

軽量イベントフック（サブスクライバの閉じ込め）

def make_emitter():
    listeners = []
    def on(handler):
        listeners.append(handler)
    def emit(value):
        for h in listeners:
            h(value)
    return on, emit

on, emit = make_emitter()
on(lambda v: print(f"A:{v}"))
on(lambda v: print(f"B:{v}"))
emit(42)
# 実行結果:
# A:42
# B:42

デコレータ

デコレータとは、関数（またはメソッド、クラス）を引数に取り、加工後の新しい関数（またはクラス）を返す高階関数である。横断的関心事（ログ、計測、バリデーション、リトライ、権限制御など）を元の実装へ非侵襲に付与するための仕組みである。
Pythonでは @decorator 構文は糖衣構文であり、@d を f の直前に書くことは概念的に f = d(f) と同値である。

仕組みの要点

・デコレータは呼び出し可能（callable）であればよい（関数でもクラスでも可）
・デコレータは元の関数を受け取ってラッパー関数（wrapper）を返す
・ラッパーは引数・戻り値・例外を適切に透過させる必要がある

基本構文

def trace(func):
    def wrapper(*args, **kwargs):
        print(f"[CALL] {func.__name__} args={args} kwargs={kwargs}")
        result = func(*args, **kwargs)
        print(f"[RET ] {result}")
        return result
    return wrapper

@trace
def add(a, b):
    return a + b

print(add(2, 5))
# 実行結果:
# [CALL] add args=(2, 5) kwargs={}
# [RET ] 7
# 7

・ @trace は add = trace(add) と等価である。
・ wrapper が実際に呼び出され、前後でログを出してから結果を返すのである。

メタデータ保全：functools.wraps は原則必須

デコレータでラップすると、元の関数名・docstring・注釈・__wrapped__ 等が失われる。functools.wraps により元のメタデータを維持すべきである。

from functools import wraps

def trace(func):
    @wraps(func)  # ← これを付けるのが実務の標準である
    def wrapper(*args, **kwargs):
        print(f"[CALL] {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

@trace
def f(x: int) -> int:
    """二乗して返す関数である。"""
    return x * x

print(f.__name__)      # 実行結果: f
print(f.__doc__)       # 実行結果: 二乗して返す関数である。
print(f.__wrapped__(3))# 実行結果: 9  （元関数へアクセス可能）

引数付きデコレータ(デコレ―タ・ファクトリ)

「デコレータ自身に設定値を渡したい」場合は、外側にもう一段階関数を作る（＝デコレータを返す関数）である。

from functools import wraps

def tag(prefix: str = "INFO"):
    def decorate(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            print(f"[{prefix}] {func.__name__} called")
            return func(*args, **kwargs)
        return wrapper
    return decorate

@tag("DEBUG")
def calc(a, b):
    return a * b

print(calc(3, 4))
# 実行結果:
# [DEBUG] calc called
# 12

多重摘要の実行順序

@A と @B を同一関数へ付けるとき、適用順は下→上（呼び出し時は上→下）である。

from functools import wraps

def A(func):
    @wraps(func)
    def w(*a, **k):
        print("A: before")
        r = func(*a, **k)
        print("A: after")
        return r
    return w

def B(func):
    @wraps(func)
    def w(*a, **k):
        print("B: before")
        r = func(*a, **k)
        print("B: after")
        return r
    return w

@A
@B
def foo():
    print("foo body")

foo()
# 実行結果:
# A: before
# B: before
# foo body
# B: after
# A: after

定義時は foo = A(B(foo)) の順でラップされるため、呼び出し時は A→B→本体→B→A の順になる。

メソッドへの適用(self とデスクリプタの要件)

関数デコレータはインスタンスメソッドにもそのまま使える。self は呼び出し時に通常どおり最初の位置に渡る。

from functools import wraps

def trace(func):
    @wraps(func)
    def w(*args, **kwargs):
        print(f"[CALL] {func.__qualname__}")
        return func(*args, **kwargs)
    return w

class Acc:
    def __init__(self): self.v = 0

    @trace
    def add(self, x):
        self.v += x
        return self.v

a = Acc()
print(a.add(10))  # 実行結果:
# [CALL] Acc.add
# 10

@classmethod / @staticmethod との順序

一般に 「関数デコレータ」→ @classmethod / @staticmethod の順で付与するか、classmethod / staticmethod のあとに関数デコレータを適用する場合はクラス宣言外でラップする。混乱を避けるため、まず関数デコレータを適用し、その後メソッド種別を付与するのが分かりやすい。

クラスベースデコレ―タ

__call__ を持つクラスをデコレ―タとして用いることで、状態fulな設定を持ち込める。

from functools import wraps

class LimitCalls:
    def __init__(self, max_calls: int = 3):
        self.max_calls = max_calls
        self.count = 0

    def __call__(self, func):
        @wraps(func)
        def w(*args, **kwargs):
            if self.count >= self.max_calls:
                raise RuntimeError("Call limit exceeded")
            self.count += 1
            return func(*args, **kwargs)
        return w

@LimitCalls(2)
def greet(name):
    return f"Hello, {name}"

print(greet("A"))  # 実行結果: Hello, A
print(greet("B"))  # 実行結果: Hello, B
try:
    print(greet("C"))
except RuntimeError as e:
    print(type(e).__name__)  # 実行結果: RuntimeError

非同期関数（async def）の装飾

対象がコルーチン関数かどうかで分岐して適切に await する。

import asyncio, inspect
from functools import wraps

def maybe_async(deco_logic):
    def decorate(func):
        if inspect.iscoroutinefunction(func):
            @wraps(func)
            async def async_w(*args, **kwargs):
                return await deco_logic(func, *args, **kwargs)
            return async_w
        else:
            @wraps(func)
            def sync_w(*args, **kwargs):
                # deco_logic は同期でも非同期でも扱えるよう await 無し分岐を別途設計してもよい
                return deco_logic(func, *args, **kwargs)
            return sync_w
    return decorate

@maybe_async(lambda f, *a, **k: f(*a, **k))
def add(x, y): return x + y

@maybe_async(lambda f, *a, **k: f(*a, **k))
async def add_async(x, y): return x + y

print(add(2, 3))  # 実行結果: 5
print(asyncio.run(add_async(2, 3)))  # 実行結果: 5

実務で使えるレシピ集

ログ＋経過時間（最小の計測器）

import time
from functools import wraps

def timeit(func):
    @wraps(func)
    def w(*args, **kwargs):
        t0 = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            dt = (time.perf_counter() - t0) * 1000
            print(f"{func.__name__} took {dt:.2f} ms")
    return w

@timeit
def heavy(n: int) -> int:
    s = 0
    for i in range(n):
        s += i*i
    return s

print(heavy(10000))
# 実行結果例:
# heavy took 0.XX ms
# 333283335000

リトライ（指数バックオフ）

import time
from functools import wraps

def retry(max_retries=3, backoff=0.1, exceptions=(Exception,)):
    def decorate(func):
        @wraps(func)
        def w(*args, **kwargs):
            delay = backoff
            for attempt in range(1, max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_retries:
                        raise
                    time.sleep(delay)
                    delay *= 2
        return w
    return decorate

i = {"n": 0}

@retry(max_retries=3, backoff=0.01, exceptions=(RuntimeError,))
def flaky():
    i["n"] += 1
    print("try", i["n"])
    raise RuntimeError("temporary")

try:
    flaky()
except RuntimeError as e:
    print("failed")  # 実行結果例: try 1 \n try 2 \n try 3 \n failed

シンプル・キャッシュ（最後の引数だけメモ化）

実務では functools.lru_cache を推奨であるが、学習のため最小例を示す。

from functools import wraps

def memoize_last_result(func):
    last_args = None
    last_result = None

    @wraps(func)
    def w(*args):
        nonlocal last_args, last_result
        if args == last_args:
            return last_result
        last_args = args
        last_result = func(*args)
        return last_result
    return w

calls = {"n": 0}

@memoize_last_result
def expensive(x):
    calls["n"] += 1
    return x * x

print(expensive(9))   # 実行結果: 81
print(expensive(9))   # 実行結果: 81 （計算スキップ）
print(calls["n"])     # 実行結果: 1

よくある落とし穴とその対策

・ wraps を忘れる：__name__ や __doc__ が失われ、inspect / テスト / ドキュメント生成が壊れる。必ず @wraps を使うべきである。
・引数・戻り値の"契約"を壊す：デコレータが対象のインタフェースを変更すると呼び出し側が破綻する。シグネチャ互換を守るべきである。必要なら inspect.signature で検査する。
・例外を握り潰す：原因究明が困難になる。捕捉する例外を限定し、ロギング後に再送出する設計が望ましい。
・状態の持ち過ぎ：デコレータが過度な状態を抱えると可観測性が下がる。設定は引数経由、状態は計測カウンタ程度に限定する。大きな状態はクラスや外部ストアへ。
・async 非対応：同期専用のデコレータを非同期関数へ適用すると await 漏れとなる。コルーチンかどうかを判別してラッパを分岐すべきである。
・順序依存の混乱：多重適用時の順序は下から適用、上から実行である。副作用の順序を設計時に明示すること。
・標準で足りる場合は再発明しない：キャッシュは functools.lru_cache、合成は functools.partial／operator、プロパティキャッシュは functools.cached_property を優先するのがよい。

ジェネレーター(yield 式)

定義

・yield は 関数実行を一時停止し、値を外へ渡しつつ関数の状態（フレーム）を保持するための式である。
・yield を内部に含む関数は、通常の関数ではなく ジェネレーター関数 になり、呼び出し時に即時実行されず ジェネレーター・オブジェクト を返す。
・以後の実行は、反復（for など）または next() / .send() によって 再開→停止 を繰り返す。

def count_up_to(n):
    i = 1
    while i <= n:
        yield i          # 値を1つ返して停止
        i += 1

g = count_up_to(3)       # ここでは実行されない（ジェネレーターが返るだけ）
list(g)                  # [1, 2, 3]

return との本質的な違い

・return： 即時に関数を終了し、以後の状態は失われる。
・yield： 関数を終了せずに 「いったん外へ値を出す」。次に呼ばれると直後の行から再開できる。
・ジェネレーター関数内の裸の return は StopIteration を送出して 反復を終了させる。
・return value（値付き return）は、StopIteration(value) の value に格納される（for では取得しないが、itertools による連携や明示的制御で参照可能）。

反復プロトコルとの関係

・for x in gen: は内部的に iter(gen) → next() を繰り返し、StopIteration を捕捉して終了する。
・この仕組みにより、大規模データでも O(1) 近い追加メモリで逐次処理ができる（ストリーミング処理）。

yield 式

Python3では yield は式であり、右辺として使える。ただし 括弧の必要がある場面に注意（内包表記や lambda は不可）。

def gen():
    x = (yield 10)      # 10 を外へ渡して停止し、次回 .send(値) が x に入る
    yield x

g = gen()
next(g)                # 10 を受け取って停止
g.send("hello")        # 'hello' を x に渡して、次の yield で 'hello' を返す
# => 'hello'

双方向通信：.send(), .throw(), .close()

・.send(value)
→ 停止中の yield 式に value を「注入」し、その yield 式の値として評価される。コルーチン的な利用が可能。
・.throw(exc)
→ 停止位置に例外を送出する。ジェネレーター側で try / except すれば握りつぶしたり別の値を yield したりできる。
・.close()
GeneratorExit を送出してクリーンアップさせ、以後の利用を終了する。

def accumulator():
    total = 0
    try:
        while True:
            x = (yield total)   # 現在の合計を返しつつ待機
            total += x
    except GeneratorExit:
        # 終了時の後始末（ログ等）
        pass

g = accumulator()
next(g)             # 0
g.send(10)          # 10
g.send(5)           # 15
g.close()           # 安全に終了

yield from による委譲

・サブジェネレーター（または任意の反復可能）へ 反復と双方向通信を丸ごと委譲する構文である。
・.send() による値伝搬、return 値の受け取り、例外伝搬などを自動で肩代わりするため、入れ子のジェネレーターを簡潔に書ける。

def sub():
    yield 1
    yield 2
    return "done"      # StopIteration("done")

def main():
    result = yield from sub()   # 1,2 をそのまま外へ流し、終了時に "result" に "done"
    yield result

list(main())  # [1, 2, 'done']

代表的な設計パターン

遅延・逐次処理（ストリーミング）

def read_lines(path):
    with open(path, "rt", encoding="utf-8") as f:
        for line in f:
            yield line.rstrip("\n")  # 1行ずつ遅延供給

メリット: ファイルが巨大でも一定メモリで処理可能。

フィルタ・マップのパイプライン化

def numbers():
    i = 0
    while True:
        i += 1
        yield i

def evens(src):
    for x in src:
        if x % 2 == 0:
            yield x

def squared(src):
    for x in src:
        yield x * x

import itertools as it
stream = squared(evens(numbers()))
list(it.islice(stream, 5))  # [4, 16, 36, 64, 100]

コルーチン（send を使う消費者）

def sink(prefix="> "):
    try:
        while True:
            item = (yield)           # 値の受け取り専用（初回は next で開始）
            print(prefix, item)
    except GeneratorExit:
        pass

s = sink("[LOG]")
next(s)               # priming（開始）
s.send("hello")       # [LOG] hello
s.send("world")       # [LOG] world
s.close()

例外と後始末

try / finally はジェネレーターでも有効。for の途中終了や .close() によって finally が必ず実行されるため、リソース解放・フラッシュなどを安全に行える。

def managed():
    print("open")
    try:
        yield 1
        yield 2
    finally:
        print("close")

for x in managed():
    print(x)
    break
# 出力:
# open
# 1
# close   ← break でも後始末が走る

パフォーマンスとメモリ特性

・ジェネレーターは 必要なときに1要素ずつ生成するため、追加メモリほぼゼロで長い列を扱える。
・ただし 要素へランダムアクセス不可、再利用には再生成が必要。
・要素計算が重い場合、yield の境界が パイプラインの粒度になり、背圧（backpressure） の自然な制御点として働く。

よくある落とし穴とその対策

・ジェネレーター式に yield は書けない
→ 内包表記やジェネレーター式の内部では yield は禁止（構文上の制約）。
・return の値は for から見えない
→ 値は StopIteration.value に入り、通常の反復文では取得しない。必要なら明示的に next() で回して捕捉するか、yield from で親に返す。
・初回 .send(value) は不可
→ 最初の停止点がないため、初回は next(gen) または gen.send(None) で プライミングが必要。
・再入不可
→ 1つのジェネレーターを並行に再入実行することはできない（同時に next() を重ねて呼ぶ等）。
・例外伝搬の理解不足
→ 外部の .throw() や内部の raise で停止状態へ例外を注入・伝搬できる。try / except の設計を明確にすること。

実務サンプルコード

チャンク分割

def chunked(iterable, size):
    it = iter(iterable)
    while True:
        buf = []
        for _ in range(size):
            try:
                buf.append(next(it))
            except StopIteration:
                break
        if not buf:
            return
        yield buf

list(chunked(range(10), 3))  # [[0,1,2],[3,4,5],[6,7,8],[9]]

例外を飲み込まず通知（ログ・継続）

def safe_map(func, xs):
    for i, x in enumerate(xs):
        try:
            yield func(x)
        except Exception as e:
            yield {"index": i, "error": repr(e)}  # 失敗も要素として流す

list(safe_map(lambda z: 100//z, [5, 0, 4]))
# [20, {'index':1,'error':'ZeroDivisionError(...)'}, 25]

yield from で段階的処理

def gen_lines(path):
    with open(path, encoding="utf-8") as f:
        yield from (line.strip() for line in f)  # 反復可能からの委譲

def filtered_words(path):
    for line in gen_lines(path):
        if line and not line.startswith("#"):
            yield from line.split()

いつ yield / ジェネレーターを用いるべきか？

・入力が大きく 全件の同時保持が不要なとき
・ストリーム処理（ファイル・ソケット・逐次API）
・パイプラインを段階化し、各段で遅延処理したいとき
・双方向制御（send によるコルーチン的やり取り）が有用なとき
・ただし、小規模データや 単回の単純処理では通常のリスト内包・関数戻り値で十分なことも多い。

参考文献

[1] 独習Python 第2版 (2025, 山田祥寛, 翔泳社)
[2] Pythonクイックリファレンス 第4版（2024, Alex, O’Reilly Japan）