Contents
スポンサードリンク
デコレータの基本と@構文
デコレータは関数を受け取り別の関数を返す高階関数です。
@ 構文は f = d(f) と同義で、構文上の可読性を提供します。
ここでは基本的な実装例と、print を使わない logging による運用上の注意を示します。
高階関数としての仕組み
デコレータは元関数をラップして前後処理を差し込みます。functools.wraps を使いメタデータを保持するのが基本です。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
from functools import wraps import logging logger = logging.getLogger(__name__) def log_calls(logger: logging.Logger | None = None, level: int = logging.INFO): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): lg = logger or logging.getLogger(func.__module__) lg.log(level, "call: %s args=%s kwargs=%s", func.__name__, args, kwargs) return func(*args, **kwargs) return wrapper return decorator # 使い方 @log_calls() def add(a, b): return a + b |
この形は運用でのログレベルやハンドラを制御しやすくします。print を直接用いないことが重要です。
実行時間計測デコレータ（副作用をオプション化）
計測結果は logger に出力し、オプション化して副作用を制御します。非同期関数に対しては装飾時の判定と実行時のフォールバックを組み合わせます。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 |
import time import inspect from functools import wraps import logging def timed(unit: str = "s", logger: logging.Logger | None = None, level: int = logging.INFO): def decorator(func): root = inspect.unwrap(func) if inspect.iscoroutinefunction(root): @wraps(func) async def async_wrapper(*args, **kwargs): lg = logger or logging.getLogger(func.__module__) start = time.perf_counter() try: return await func(*args, **kwargs) finally: elapsed = time.perf_counter() - start if unit == "ms": elapsed *= 1000 lg.log(level, "%s took %.6f%s", func.__name__, elapsed, unit) return async_wrapper else: @wraps(func) def sync_wrapper(*args, **kwargs): lg = logger or logging.getLogger(func.__module__) start = time.perf_counter() try: return func(*args, **kwargs) finally: elapsed = time.perf_counter() - start if unit == "ms": elapsed *= 1000 lg.log(level, "%s took %.6f%s", func.__name__, elapsed, unit) return sync_wrapper return decorator # 使い方（同期） @timed(unit="ms") def work(n: int) -> int: return sum(range(n)) |
副作用は明示的に設定し、計測コードが例外を握りつぶさないようにしてください。
デコレータのメタデータ保持と元関数の安全取得
functools.wraps は name や doc、annotations を引き継ぎますが、元関数取得には inspect.unwrap を使うべきです。
wrapped に直接依存すると多重デコレータや存在しない場合に失敗します。inspect.unwrap は安全で堅牢です。
functools.wraps と wrapped の注意点
wraps は wrapped を付与しますが、直接参照する手法は脆弱です。inspect.unwrap を使えば多層ラップでも元関数を安全に取得できます。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
from functools import wraps import inspect def preserve(func): @wraps(func) def wrapper(*a, **kw): return func(*a, **kw) return wrapper @preserve def f(x: int) -> int: return x + 1 # 正しい元関数の取得 root = inspect.unwrap(f) sig = inspect.signature(root) |
テストやモックで元のシグネチャを参照する際は inspect.unwrap を使ってください。
wrapt を使う利点
高度な署名保持や複雑なラップを行う場合は wrapt ライブラリが有用です。wrapt はラッパーの実装を安全に行うための補助を提供します。小規模なケースは標準ライブラリで十分ですが、ライブラリ提供者や複雑なフローには検討してください。
非同期対応を含むデコレータ設計
非同期関数を正しく扱うことは実務で必須です。装飾時だけで同期/非同期を判定するのは不完全なことがあります。inspect.unwrap を用いた装飾時判定と、実行時の awaitable 判定の組み合わせが安定します。
装飾時の同期/非同期判定と限界
inspect.iscoroutinefunction は便利ですが、functools.partial や C 拡張、既にラップされた関数では誤判定することがあります。装飾時には inspect.unwrap でルート関数を確認してください。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
import inspect from functools import wraps def maybe_async_deco(func): root = inspect.unwrap(func) if inspect.iscoroutinefunction(root): @wraps(func) async def async_wrapper(*args, **kwargs): # async 用の処理 return await func(*args, **kwargs) return async_wrapper # type: ignore else: @wraps(func) def sync_wrapper(*args, **kwargs): # sync 用の処理 result = func(*args, **kwargs) # 実行時に awaitable が返る場合はそのまま返す return result return sync_wrapper |
型チェック器への対応は難しいため、必要なら overload を使うか type: ignore を注記してください。装飾時判定だけに頼らない設計が重要です。
実行時の awaitable 判定によるフォールバック
装飾時に非同期と判定できなかったケースでも、呼び出し時に結果が awaitable かを判定することで安全に扱えます。非同期装飾済みのオブジェクトが awaitable を返す場合などに対応できます。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
import inspect from functools import wraps def universal_deco(func): root = inspect.unwrap(func) if inspect.iscoroutinefunction(root): @wraps(func) async def async_wrapper(*args, **kwargs): result = func(*args, **kwargs) if inspect.isawaitable(result): return await result return result return async_wrapper else: @wraps(func) def sync_wrapper(*args, **kwargs): # sync ラッパーは await しない。呼び出し側が await する設計を尊重する。 return func(*args, **kwargs) return sync_wrapper |
呼び出し側の期待（同期なら同期、非同期なら await が必要）を壊さない設計を優先してください。
インスタンス単位キャッシュを安全に実装するデコレータ
インスタンスごとのキャッシュは便利ですが、スレッド競合やキー生成の扱いに注意が必要です。ここでは競合を避けるロック設計と堅牢なキー生成を示します。
問題点の整理
ここで想定する主な問題点は次のとおりです。
- ロックの範囲が不適切だとレースが発生する。
- args/kwargs をそのままキーにすると非ハッシュ可能な値で TypeError が発生する。
- 複数スレッドが同時に同じ計算を始めると重複計算や不整合が起きる。
これらを防ぐ設計を導入します。
堅牢なキー生成の方針
キーは可能な限り再現可能でハッシュ可能な表現に変換します。基本方針は次の通りです。
- 基本型はそのまま使う。
- リストや集合、辞書は再帰的にタプル化してソートする。
- ハッシュ化できないオブジェクトは repr() や id() をフォールバックに使う（副作用と衝突の可能性を説明して運用する）。
以下は「ベストエフォート」で変換するユーティリティの例です。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
import types def _make_hashable(obj): if obj is None or isinstance(obj, (int, float, str, bytes, bool)): return obj if isinstance(obj, (tuple, list)): return tuple(_make_hashable(x) for x in obj) if isinstance(obj, dict): return tuple(sorted((_make_hashable(k), _make_hashable(v)) for k, v in obj.items())) if isinstance(obj, set): return tuple(sorted(_make_hashable(x) for x in obj)) # 最後は hash できるか試し、だめなら repr や id にフォールバック try: return ('hash', hash(obj)) except Exception: try: return ('repr', repr(obj)) except Exception: return ('id', id(obj), type(obj).__name__) def _make_key(args, kwargs): return (_make_hashable(args), _make_hashable(kwargs)) |
完全な衝突回避は難しいため、設計時に用途とトレードオフを評価してください。
スレッド安全なロック設計（実装例）
以下は per-instance キャッシュの簡易実装例です。要点は次の通りです。
- WeakKeyDictionary を使いインスタンスに紐づくデータを保持する。
- グローバルな作成/取得は短時間で済ませるために専用のロックで保護する。
- 各インスタンスごとにキャッシュと「inflight」制御を持ち、同じキーの同時計算を同期的に待機できるようにする。
- 非同期関数には asyncio の primitives を使って同様の制御を行う。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
import weakref import threading import asyncio import inspect from functools import wraps class _SyncEvent(threading.Event): def __init__(self): super().__init__() self.result = None self.exc = None def per_instance_cache(key_maker=_make_key): def decorator(func): data = weakref.WeakKeyDictionary() data_lock = threading.RLock() root = inspect.unwrap(func) if inspect.iscoroutinefunction(root): async def async_wrapper(self_obj, *args, **kwargs): with data_lock: entry = data.get(self_obj) if entry is None: entry = {'cache': {}, 'inflight': {}, 'lock': asyncio.Lock()} data[self_obj] = entry key = key_maker(args, kwargs) async with entry['lock']: if key in entry['cache']: return entry['cache'][key] if key in entry['inflight']: fut = entry['inflight'][key] compute = False else: fut = asyncio.get_running_loop().create_future() entry['inflight'][key] = fut compute = True if not compute: return await fut try: result = await func(self_obj, *args, **kwargs) except Exception as e: async with entry['lock']: fut = entry['inflight'].pop(key) fut.set_exception(e) raise async with entry['lock']: entry['cache'][key] = result fut = entry['inflight'].pop(key) fut.set_result(result) return result return wraps(func)(async_wrapper) else: def sync_wrapper(self_obj, *args, **kwargs): with data_lock: entry = data.get(self_obj) if entry is None: entry = {'cache': {}, 'inflight': {}, 'lock': threading.RLock()} data[self_obj] = entry key = key_maker(args, kwargs) with entry['lock']: if key in entry['cache']: return entry['cache'][key] if key in entry['inflight']: ev = entry['inflight'][key] compute = False else: ev = _SyncEvent() entry['inflight'][key] = ev compute = True if not compute: ev.wait() with entry['lock']: if ev.exc: raise ev.exc return entry['cache'][key] try: val = func(self_obj, *args, **kwargs) except Exception as e: with entry['lock']: ev = entry['inflight'].pop(key) ev.exc = e ev.set() raise with entry['lock']: entry['cache'][key] = val ev = entry['inflight'].pop(key) ev.result = val ev.set() return val return wraps(func)(sync_wrapper) return decorator |
運用上の注意点は次の通りです。キー生成の戦略が衝突や一貫性に与える影響、キャッシュのクリア方法、メモリ使用量の監視です。大規模運用では標準の cached_property や lru_cache、専用ライブラリを優先してください。
実務ユースケース別のデコレータ実装と外部ライブラリ選定
ここではよく使うパターンと運用上の注意を示します。各例は logging を使い副作用をコントロールします。外部ライブラリは公式ドキュメントと信頼できる記事を優先して評価してください。
純関数向けキャッシュ（functools.lru_cache）
純関数には functools.lru_cache を優先してください。安全で最適化済みです。
|
1 2 3 4 5 6 7 8 |
from functools import lru_cache @lru_cache(maxsize=128) def fib(n: int) -> int: if n < 2: return n return fib(n-1) + fib(n-2) |
cached_property（Python 3.8+）
インスタンス属性の遅延評価には functools.cached_property を使ってください。
|
1 2 3 4 5 6 7 |
from functools import cached_property class Expensive: @cached_property def value(self): return sum(range(10_000)) |
リトライ／バックオフ（同期／非同期）
汎用的な retry は exceptions 引数で捕捉対象を限定してください。デフォルトで BaseException 系は捕まえないようにします。tenacity の利用を検討してください。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 |
import time import asyncio import inspect from functools import wraps def retry(retries=3, delay=0.1, exceptions=(Exception,)): def decorator(func): root = inspect.unwrap(func) if inspect.iscoroutinefunction(root): @wraps(func) async def async_wrapper(*args, **kwargs): last_exc = None for i in range(retries): try: return await func(*args, **kwargs) except exceptions as e: last_exc = e if i + 1 < retries: await asyncio.sleep(delay * (2 ** i)) else: raise return async_wrapper else: @wraps(func) def sync_wrapper(*args, **kwargs): last_exc = None for i in range(retries): try: return func(*args, **kwargs) except exceptions as e: last_exc = e if i + 1 < retries: time.sleep(delay * (2 ** i)) else: raise return sync_wrapper return decorator |
認可とロギング
認可チェックは例外で失敗を明示し、ログは logging を使います。フレームワーク依存のコードは抽象化すると差し替えやすくなります。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
from functools import wraps import logging logger = logging.getLogger(__name__) def login_required(check): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): if not check(): logger.warning("unauthorized access to %s", func.__name__) raise PermissionError("not logged in") return func(*args, **kwargs) return wrapper return decorator |
入力検証
軽量な検証はデコレータで行えますが、複雑な検証は pydantic 等を使うと保守性が高まります。ランタイムコストを評価してください。
外部ライブラリの選定（優先順）
- 標準ライブラリ: functools, inspect, asyncio をまず検討します。
- wrapt: 署名保持と複雑なラップに強いです。
- tenacity: 再試行とバックオフに特化し、sync/async 両対応です。
- pydantic/typeguard: ランタイム検証が必要な場合に検討します。
- 参考は Python の公式ドキュメントや関連 PEP を優先してください。Real Python の解説記事も参考になります。
デコレータのまとめ
デコレータは横断的処理を簡潔に実装できますが、運用上の設計が重要です。
特にメタデータ保持、同期/非同期の判定、スレッド安全なキャッシュ、キー生成の堅牢化に注意してください。
以下は本記事の要点です。
- functools.wraps と inspect.unwrap を組み合わせてメタデータと元関数を安全に扱う。
- inspect.iscoroutinefunction は便利だが限界があるため、装飾時判定と実行時の awaitable 判定を併用する。
- per-instance キャッシュは WeakKeyDictionary と per-instance ロック、inflight 制御で競合を避ける。
- キー生成は再帰的にハッシュ可能な形へ変換し、最終的には repr/id にフォールバックする方針を採る。
- print を直接使わず logging を使い、ログレベルやハンドラを運用で制御する。
- tenacity や wrapt 等、信頼できる外部ライブラリは必要に応じて導入する。
参考（抜粋）:
- Python 標準ドキュメント（functools, inspect, asyncio）
- PEP（関連する仕様、ParamSpec に関する PEP など）
- Real Python のデコレータ入門や wrapt のドキュメント
以上の原則を踏まえ、まずは小さなユーティリティから導入し、ユニットテストで sync/async の両パスを必ずカバーしてください。
スポンサードリンク