Skip to the content.

← 目次← 前: 24次: 26 →

25. デコレータと関数の高階操作

テーマ: 実行時間計測・メモ化・リトライ・入力検証をデコレータで実装する

学習点: 関数はオブジェクト, クロージャ, functools.wraps / lru_cache, 引数付きデコレータ, スタック(積み重ね)

依存: 標準ライブラリのみ / 難易度: 上級

実行方法

uv run 25_decorators.py

スクリプト冒頭の PEP 723 メタデータ(# /// script)により、必要なライブラリは uv が自動的に仮想環境へ導入します。事前の pip install は不要です。

解説

何をするプログラムか

実行時間の計測、計算結果のキャッシュ(メモ化)、外部 API 呼び出しの再試行、引数の検証――こうした処理は業務システムのあらゆる関数に共通して必要になる「横断的関心事」です。それらを本体のロジックに書き込むとコードが汚れます。Python のデコレータは、既存の関数を包んで機能を後付けする仕組みで、この問題をエレガントに解決します。

このスクリプトは timermemoizeretryvalidate_positive という 4 つの実用デコレータを自作し、フィボナッチ数列や NPV 計算、不安定な API 呼び出しの模擬に適用します。デコレータが成立する土台は「関数はオブジェクトであり、変数に代入したり関数へ渡したりできる」という Python の性質そのものです。番号付きの 5 つの実験を通して、構文糖衣の正体からクラスベースのデコレータまで段階的に学べます。

コードの読みどころ

実行結果の見方

[1] が最大の見どころです。素朴な再帰は F(32) に 417.877 ms かかるのに対し、自前メモ化は 0.029 ms、lru_cache は 0.012 ms と 1 万倍以上高速です。素朴な再帰が約 436 万回の呼び出しを行う一方、CacheInfo(hits=31, misses=33, ...) が示すとおり、メモ化版は実際の計算(miss)を 33 回しか行っていません。

[2] では負の割引率を渡した npv(-0.05, 100)validate_positive に阻まれ、検証エラーになります。[3] のリトライは今回 1 回目で成功したため失敗ログが出ていません(random.seed(0) の最初の乱数が失敗率 0.7 を上回ったためで、失敗すると [retry] n回目 失敗: の行が現れます)。[5] は 3 回のループ後に compound.count = 3 となり、デコレータが呼び出し回数という状態を保持できることを示しています。

ソースコード

# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""25: デコレータと関数の高階操作 -----------------------------------------
テーマ: 実行時間計測・メモ化・リトライ・入力検証をデコレータで実装する
学習点: 関数はオブジェクト, クロージャ, functools.wraps / lru_cache,
        引数付きデコレータ, スタック(積み重ね)
"""
import functools
import random
import time

random.seed(0)


def timer(func):
    """実行時間を計測して表示するデコレータ。"""
    @functools.wraps(func)          # メタデータ(__name__等)を保持する
    def wrapper(*args, **kwargs):
        t0 = time.perf_counter()
        result = func(*args, **kwargs)
        print(f"    [timer] {func.__name__}: "
              f"{(time.perf_counter() - t0) * 1000:.3f} ms")
        return result
    return wrapper


def memoize(func):
    """自前のメモ化(キャッシュ)。辞書に計算結果を溜める。"""
    cache = {}

    @functools.wraps(func)
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    wrapper.cache = cache
    return wrapper


def retry(times: int = 3, delay: float = 0.01):
    """引数を取るデコレータ(デコレータを返す関数)。"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"    [retry] {attempt}回目 失敗: {e}")
                    if attempt == times:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator


def validate_positive(func):
    """引数がすべて正であることを検証する。"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        for a in args:
            if isinstance(a, (int, float)) and a <= 0:
                raise ValueError(f"{func.__name__}: 正の数が必要です (got {a})")
        return func(*args, **kwargs)
    return wrapper


# ---- 適用例 ----
def fib_naive(n: int) -> int:
    return n if n < 2 else fib_naive(n - 1) + fib_naive(n - 2)


@memoize
def fib_memo(n: int) -> int:
    return n if n < 2 else fib_memo(n - 1) + fib_memo(n - 2)


@functools.lru_cache(maxsize=None)
def fib_lru(n: int) -> int:
    return n if n < 2 else fib_lru(n - 1) + fib_lru(n - 2)


@timer
@validate_positive
def npv(rate: float, *cashflows: float) -> float:
    return sum(cf / (1 + rate) ** t for t, cf in enumerate(cashflows))


@retry(times=4)
def unstable_api_call(fail_rate: float = 0.7) -> str:
    if random.random() < fail_rate:
        raise ConnectionError("外部APIに接続できません")
    return "データ取得成功"


def main() -> None:
    print("[1] メモ化の効果 — フィボナッチ数列 F(32)")
    for name, f in [("素朴な再帰", fib_naive), ("自前メモ化", fib_memo),
                    ("lru_cache", fib_lru)]:
        t0 = time.perf_counter()
        v = f(32)
        dt = (time.perf_counter() - t0) * 1000
        print(f"  {name:<12} F(32) = {v:>10,}  {dt:>10.3f} ms")
    print(f"  素朴な再帰の呼び出し回数は O(φ^n) ≈ {2 * fib_lru(32):,} 回相当")
    print(f"  lru_cache の統計: {fib_lru.cache_info()}")
    print(f"  自前キャッシュのエントリ数: {len(fib_memo.cache)}\n")

    print("[2] デコレータのスタックと検証")
    print(f"  NPV = {npv(0.08, 1000, 400, 400, 400):.4f}")
    try:
        npv(-0.05, 100)
    except ValueError as e:
        print(f"  検証エラー: {e}")
    print(f"  functools.wraps のおかげで __name__ は保持される: "
          f"{npv.__name__}\n")

    print("[3] リトライ")
    print(f"  結果: {unstable_api_call()}\n")

    print("[4] デコレータの正体 — 構文糖衣の展開")
    print("  @timer            timer(f) を f に再代入しているだけ")
    print("  def f(): ...   ⇔  f = timer(f)")
    plain = lambda x: x ** 2
    print(f"  手動適用: {timer(plain)(12) = }")

    print("\n[5] クラスベースのデコレータ(呼び出し回数を数える)")

    class CountCalls:
        def __init__(self, func):
            functools.update_wrapper(self, func)
            self.func = func
            self.count = 0

        def __call__(self, *args, **kwargs):
            self.count += 1
            return self.func(*args, **kwargs)

    @CountCalls
    def compound(pv, r, n):
        return pv * (1 + r) ** n

    for r in [0.01, 0.02, 0.03]:
        compound(100, r, 10)
    print(f"  compound の呼び出し回数 = {compound.count}")


if __name__ == "__main__":
    main()

実行結果

[1] メモ化の効果 — フィボナッチ数列 F(32)
  素朴な再帰        F(32) =  2,178,309     417.877 ms
  自前メモ化        F(32) =  2,178,309       0.029 ms
  lru_cache    F(32) =  2,178,309       0.012 ms
  素朴な再帰の呼び出し回数は O(φ^n) ≈ 4,356,618 回相当
  lru_cache の統計: CacheInfo(hits=31, misses=33, maxsize=None, currsize=33)
  自前キャッシュのエントリ数: 33

[2] デコレータのスタックと検証
    [timer] npv: 0.014 ms
  NPV = 2030.8388
  検証エラー: npv: 正の数が必要です (got -0.05)
  functools.wraps のおかげで __name__ は保持される: npv

[3] リトライ
  結果: データ取得成功

[4] デコレータの正体 — 構文糖衣の展開
  @timer            timer(f) を f に再代入しているだけ
  def f(): ...   ⇔  f = timer(f)
    [timer] <lambda>: 0.002 ms
  手動適用: timer(plain)(12) = 144

[5] クラスベースのデコレータ(呼び出し回数を数える)
  compound の呼び出し回数 = 3

← 目次← 前: 24次: 26 →