Contents
1️⃣ デコレータの基本概念と導入メリット
📌 ポイント
- デコレータ は「関数(またはクラス)に前後処理を付加」する構文糖です。
- ロギング、認証、キャッシュ、リトライなど 横断的関心事 を個別実装から切り離し、再利用可能 な部品として管理できます。
🎯 なぜ必要か
大規模プロジェクトでは同じ処理が多数の関数に散在します。
- 保守性低下 → 変更漏れ・バグ増加
- 可読性低下 → ビジネスロジックとインフラ系コードが混在
デコレータを導入すれば、横断的処理は 1 カ所 に集約でき、テストやログの一元管理も容易になります。
🔧 シンプルなロギング例(インポート・コメント付き)
|
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 |
# ---------------------------------------------- # ① 必要モジュールをまとめてインポート # ---------------------------------------------- import functools # デコレータ作成時は必ず使用 import logging # 標準ロガー import time # タイミング計測に利用 # ロガー設定(実務では config ファイルで管理) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ---------------------------------------------- # ② デコレータ本体 # ---------------------------------------------- def log_execution(func): """ 関数呼び出しの開始・終了を INFO レベルで記録するデコレータ。 """ @functools.wraps(func) # 元関数の __name__ 等を保持 def wrapper(*args, **kwargs): logger.info(f"CALL {func.__name__} args={args} kwargs={kwargs}") result = func(*args, **kwargs) logger.info(f"RETURN {func.__name__} -> {result!r}") return result return wrapper # ---------------------------------------------- # ③ デコレータ適用例 # ---------------------------------------------- @log_execution def process_data(data): """ダミーのデータ変換処理(実務では複雑ロジック)""" transformed = data.upper() # ← 実装は任意 return transformed |
ポイント:
functools.wrapsが無いとprocess_data.__name__が"wrapper"になり、デバッグが困難になります。
2️⃣ デコレータ作成時の必須テクニック
2.1 functools.wraps とメタデータ保持
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
import functools import time def timer(func): """ 実行時間を計測し、標準出力へ表示するデコレータ。 """ @functools.wraps(func) # ← ここで __name__ 等がコピーされる def wrapper(*args, **kwargs): start = time.perf_counter() value = func(*args, **kwargs) elapsed = time.perf_counter() - start print(f"{func.__name__} took {elapsed:.4f}s") return value return wrapper |
@functools.wrapsは 必須 のマジックです。- デバッグツールや自動生成ドキュメントが正しい名前を取得できるようになります。
2.2 可変長引数・キーワード引数で汎用ラッパー
|
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 |
import functools import logging logger = logging.getLogger(__name__) def safe_execute(default=None): """ 例外発生時に `default` を返す汎用デコレータ。 """ def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logger.exception(f"{func.__name__} raised {e}") return default return wrapper return decorator @safe_execute(default={}) def load_config(path: str) -> dict: """JSON 設定ファイルをロードし、失敗時は空辞書を返す""" import json # ここで遅延インポートしても OK with open(path, encoding="utf-8") as f: return json.load(f) |
*args, **kwargsをそのまま転送することで、任意のシグネチャ に対応できます。- ロギングは
loggingモジュールを直接利用し、外部依存 (logger) が未定義になるリスクを排除しました。
2.3 複数デコレータの適用順序と可視化
|
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 |
import functools def deco_a(func): @functools.wraps(func) def wrapper(*args, **kwargs): print("A before") result = func(*args, **kwargs) print("A after") return result return wrapper def deco_b(func): @functools.wraps(func) def wrapper(*args, **kwargs): print("B before") result = func(*args, **kwargs) print("B after") return result return wrapper @deco_a # ← 外側(最後に呼ばれる) @deco_b # ← 内側(最初に呼ばれる) def hello(): """シンプルな出力だけを行う関数""" print("Hello") # 実行例 hello() |
実行結果
|
1 2 3 4 5 6 |
A before B before Hello B after A after |
| デコレータ | 適用位置 | 実行タイミング |
|---|---|---|
deco_b |
内側 | 呼び出し前 → 後 |
deco_a |
外側 | 呼び出し前 → 後 |
覚えておくべきこと:デコレータは 下から上へ(
f = d1(d2(f)))適用され、実行順序は逆になります。テーブルやコメントで可視化すれば、意図しない順序ミスを防げます。
3️⃣ クラスベースデコレータ & 型安全実装
3.1 状態保持が必要なときはクラスデコレータ
|
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 functools import time import logging logger = logging.getLogger(__name__) class Retry: """ 呼び出し失敗時にリトライするデコレータ。 - attempts: 最大リトライ回数(デフォルト 3 回) - delay: 各リトライ間の待機秒数(デフォルト 0.5 秒) """ def __init__(self, attempts: int = 3, delay: float = 0.5): self.attempts = attempts self.delay = delay def __call__(self, func): @functools.wraps(func) def wrapper(*args, **kwargs): for i in range(1, self.attempts + 1): try: return func(*args, **kwargs) except Exception as e: if i == self.attempts: raise logger.warning(f"Retry {i}/{self.attempts} after error: {e}") time.sleep(self.delay) return wrapper # 使用例 @Retry(attempts=5, delay=1) def unreliable_api(): """ランダムに例外を投げるサンプル関数(実務では外部 API 呼び出し等)""" import random if random.random() < 0.7: raise RuntimeError("Transient error") return "Success" |
- メリット:インスタンス属性で
attemptsやdelayを保持でき、関数ごとに異なる設定が容易です。
3.2 Python 3.12+ の Protocol と ParamSpec で型安全に
|
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 |
# ------------------------------------------------- # 必要な型ヒント系インポート # ------------------------------------------------- from __future__ import annotations import functools import time from typing import Protocol, Callable, TypeVar, ParamSpec, runtime_checkable P = ParamSpec('P') R = TypeVar('R') @runtime_checkable # ← これが無いと isinstance が機能しない class DecoratorProtocol(Protocol[P, R]): """ デコレータ自身が ``Callable[[Callable[..., ...]], Callable[..., ...]]`` の形であることを表す Protocol。 """ def __call__(self, func: Callable[P, R]) -> Callable[P, R]: ... def timing_decorator(func: Callable[P, R]) -> Callable[P, R]: """実行時間を測定し、標準出力に表示するシンプルデコレータ""" @functools.wraps(func) def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: start = time.perf_counter() result = func(*args, **kwargs) elapsed = time.perf_counter() - start print(f"{func.__name__} executed in {elapsed:.4f}s") return result return wrapper # 型チェック(mypy や pyright が通ることを確認) assert isinstance(timing_decorator, DecoratorProtocol) |
@runtime_checkableを付与することで、実行時にisinstance判定が可能になります。ParamSpecとTypeVarにより、ラップ前後でシグネチャが完全に保持されることを型レベルでも保証できます。
4️⃣ 実務でよく使うユースケース例(Django / Flask)
| ユースケース | 主な目的 | コードスニペット |
|---|---|---|
| ロギング | 呼び出し情報・例外を記録 | 既述 log_execution デコレータ |
| 認証/権限チェック | API キーやユーザー権限の検証 | Flask / Django 用実装 ↓ |
| キャッシュ | 計算結果や DB クエリ結果の再利用 | Django ORM キャッシュ例 ↓ |
| リトライ | 一時的失敗(ネットワーク等)の自動再試行 | Retry クラス例 |
| タイムアウト | 処理が規定時間を超えたら中断 | Flask 用 timeout デコレータ |
4.1 Flask – API キー認証デコレータ
|
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 |
import os import functools from flask import request, abort def require_api_key(func): """ リクエストヘッダーに正しい X-API-Key が含まれているか検証する。 環境変数 ``API_KEY`` と比較し、不一致なら 401 を返す。 """ @functools.wraps(func) def wrapper(*args, **kwargs): key = request.headers.get('X-API-Key') if key != os.getenv('API_KEY'): abort(401, description="Invalid API Key") return func(*args, **kwargs) return wrapper # Flask アプリ例(省略可) # from flask import Flask # app = Flask(__name__) # @app.route('/protected') # @require_api_key # def protected(): # return "You have access" |
4.2 Django – IP アドレス制限デコレータ
|
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 |
import functools from django.http import HttpResponseForbidden from django.conf import settings def ip_allowlist(allowed_ips): """ 許可された IP アドレスリストに含まれない場合は 403 を返す。 ``settings.ALLOWED_IPS`` に設定したリストを直接渡すと便利。 """ def decorator(view_func): @functools.wraps(view_func) def _wrapped_view(request, *args, **kwargs): client_ip = request.META.get('REMOTE_ADDR') if client_ip not in allowed_ips: return HttpResponseForbidden("IP not allowed") return view_func(request, *args, **kwargs) return _wrapped_view return decorator # settings.py の例 # ALLOWED_IPS = ['203.0.113.5', '198.51.100.23'] @ip_allowlist(settings.ALLOWED_IPS) def secret_view(request): """IP 制限された機密ビュー""" from django.http import HttpResponse return HttpResponse("Secret data") |
4.3 Flask – リクエストタイムアウト(signal 利用)
|
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 |
import functools import signal import time from flask import abort class TimeoutException(RuntimeError): """内部で使用する例外クラス""" def timeout(seconds: int): """ 指定秒数を超えると 504 エラーを返すデコレータ。 Unix 系 OS のみ動作(signal.SIGALRM が利用できる環境)。 """ def decorator(func): @functools.wraps(func) def _wrapped(*args, **kwargs): def _handle(signum, frame): raise TimeoutException() old_handler = signal.signal(signal.SIGALRM, _handle) signal.alarm(seconds) # アラーム開始 try: return func(*args, **kwargs) except TimeoutException: abort(504, description="Request timed out") finally: signal.alarm(0) # アラーム解除 signal.signal(signal.SIGALRM, old_handler) return _wrapped return decorator # Flask ルート例 # @app.route('/slow') # @timeout(2) # def slow_endpoint(): # time.sleep(5) # 実際は外部 API 呼び出し等 # return "Finished" |
ポイント:Flask と Django は「デコレータでリクエストオブジェクトにアクセス → HTTP レスポンスを返す」流れが共通です。公式デコレータ実装を参考に、独自ロジック(IP 制限・タイムアウト等)を組み込むだけで拡張が可能です。
5️⃣ デコレータ作成時の注意点 & 導入ステップ
✅ 注意点チェックリスト
| 項目 | 推奨対策 |
|---|---|
| 副作用回避 | グローバル変数は直接書き換えず、threading.local() や contextvars.ContextVar を利用 |
| スレッド安全性 | 共有キャッシュやリトライカウンタは Lock で保護、もしくは cachetools.TTLCache のようなスレッドセーフ実装を使用 |
| 例外ハンドリング | デコレータ内部で捕捉しすぎない。必要最小限の例外だけを処理し、残りは呼び出し元へ再送出 |
| テスト容易性 | 小さく保ち、unittest.mock でラップされた関数を差し替えて振る舞いを検証 |
| 型ヒント | Python 3.12+ の Protocol・ParamSpec を活用し、IDE 補完と static type checking を有効化 |
| ドキュメント | functools.wraps により元関数の docstring が残るが、デコレータ独自のオプションはコメントまたは別途 README で明示 |
🚀 導入ステップ(実務向け)
- 要件定義
- 対象とする横断的処理を洗い出す(例: ロギング+リトライ)。
- プロトタイプ作成
functools.wrapsと*args, **kwargsだけで簡易実装。- 型ヒント追加
Protocol・ParamSpecを入れ、mypy --strictでエラーが出ないことを確認。- ユニットテスト作成
- 正常系・例外系・マルチスレッドシナリオを網羅。
- CI/CD 組み込み
- カバレッジ ≥ 90%、静的解析(
ruff,flake8,mypy)の結果がグリーンになること。 - 本番デプロイ & モニタリング
- ログやメトリクスで呼び出し回数・例外率を可視化し、パフォーマンス影響を定量評価。
6️⃣ まとめ
| 項目 | 内容 |
|---|---|
| デコレータの意義 | 関数・クラスの振る舞いを非侵入的に拡張し、横断的処理の再利用性と保守性を向上させる |
| 必須テクニック | functools.wraps でメタデータ保持、*args, **kwargs で汎用ラッパー実装、適用順序の可視化 |
| 高度な実装 | クラスベース(状態保持)と Python 3.12+ の Protocol・ParamSpec による型安全デコレータ |
| 実務ユースケース | ロギング、認証/権限チェック、キャッシュ、リトライ、タイムアウトを Django / Flask で具体化 |
| 開発フロー | 要件 → プロトタイプ → 型ヒント → テスト → CI/CD → 本番モニタリング |
最終的に、上記のベストプラクティスとチェックリストをプロジェクトに組み込めば、デコレータはコードベース全体の可読性・信頼性を大幅に向上させる「インフラ層」の重要コンポーネントとなります。ぜひ自チームで共有し、標準実装パターンとして定着させてください。