25. デコレータと関数の高階操作
テーマ: 実行時間計測・メモ化・リトライ・入力検証をデコレータで実装する
学習点: 関数はオブジェクト, クロージャ, functools.wraps / lru_cache, 引数付きデコレータ, スタック(積み重ね)
依存: 標準ライブラリのみ / 難易度: 上級
実行方法
uv run 25_decorators.py
スクリプト冒頭の PEP 723 メタデータ(# /// script)により、必要なライブラリは
uv が自動的に仮想環境へ導入します。事前の pip install は不要です。
解説
何をするプログラムか
実行時間の計測、計算結果のキャッシュ(メモ化)、外部 API 呼び出しの再試行、引数の検証――こうした処理は業務システムのあらゆる関数に共通して必要になる「横断的関心事」です。それらを本体のロジックに書き込むとコードが汚れます。Python のデコレータは、既存の関数を包んで機能を後付けする仕組みで、この問題をエレガントに解決します。
このスクリプトは timer・memoize・retry・validate_positive という 4 つの実用デコレータを自作し、フィボナッチ数列や NPV 計算、不安定な API 呼び出しの模擬に適用します。デコレータが成立する土台は「関数はオブジェクトであり、変数に代入したり関数へ渡したりできる」という Python の性質そのものです。番号付きの 5 つの実験を通して、構文糖衣の正体からクラスベースのデコレータまで段階的に学べます。
コードの読みどころ
timer(func)は内部で定義したwrapperを返す典型的な形です。@functools.wraps(func)を付けることで__name__などのメタデータが包まれた後も保持され、[2] の出力でnpv.__name__がwrapperではなくnpvのままであることが確認できます。memoizeは外側の関数のローカル変数cache = {}をwrapperがクロージャとして抱え込みます。さらにwrapper.cache = cacheと属性を生やしているため、実行後にlen(fib_memo.cache)でエントリ数 33 を外から観察できます。retry(times=4)は引数付きデコレータで、retry → decorator → wrapperの 3 層構造です。「デコレータを返す関数」を 1 段かぶせるのがポイントで、失敗時はtime.sleep(delay)を挟んで再試行し、最終回だけraiseで例外を投げ直します。npvには@timerと@validate_positiveが積み重ねられています。デコレータは下から順に適用されるため、実体はtimer(validate_positive(npv))です。検証が先に走り、計測は検証込みの時間を測ります。- [4] では
plain = lambda x: x ** 2にtimer(plain)(12)と手動適用し、@timerが単なる「f = timer(f)の構文糖衣」であることを示します。 - [5] の
CountCallsは__call__を実装したクラスで、インスタンス自身が呼び出し可能になるためデコレータとして使えます。呼び出し回数のような状態を持たせたいときはクラスベースが向いています。
実行結果の見方
[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