Pythonデコレータ実務パターン：業務効率化の実装テクニック

def decorator(func):\n    def wrapper(*args, **kwargs):\n        # 前処理\n        result = func(*args, **kwargs)\n        # 後処理\n        return result\n    return wrapper\n\n@decorator\ndef my_function():\n    pass

import time\nimport functools\nfrom typing import Any, Callable\n\ndef measure_execution_time(func: Callable) -> Callable:\n    \"\"\"関数の実行時間を計測してログに出力するデコレータ\"\"\"\n    @functools.wraps(func)\n    def wrapper(*args, **kwargs) -> Any:\n        start_time = time.time()\n        try:\n            result = func(*args, **kwargs)\n            return result\n        finally:\n            elapsed_time = time.time() - start_time\n            print(f\"[{func.__name__}] 実行時間: {elapsed_time:.3f}秒\")\n    return wrapper\n\n@measure_execution_time\ndef fetch_user_data(user_id: int) -> dict:\n    \"\"\"ユーザーデータを取得する重い処理\"\"\"\n    time.sleep(1.5)  # 実際のAPI呼び出しを想定\n    return {\"user_id\": user_id, \"name\": \"太郎\"}\n\n# 使用例\nresult = fetch_user_data(123)\nprint(result)\n# 出力: [fetch_user_data] 実行時間: 1.502秒\n# {'user_id': 123, 'name': '太郎'}

2. リトライロジック付きデコレータ

ネットワーク処理やAPI呼び出しで失敗することがあります。自動でリトライする仕組みを実装すれば、一時的なエラーに強くなります。

import functools\nimport random\nfrom typing import Any, Callable, Type\n\ndef retry_on_exception(\n    max_attempts: int = 3,\n    delay: float = 1.0,\n    exceptions: tuple = (Exception,)\n) -> Callable:\n    \"\"\"指定した例外が発生したら自動でリトライするデコレータ\"\"\"\n    def decorator(func: Callable) -> Callable:\n        @functools.wraps(func)\n        def wrapper(*args, **kwargs) -> Any:\n            attempt = 0\n            last_exception = None\n            \n            while attempt < max_attempts:\n                try:\n                    return func(*args, **kwargs)\n                except exceptions as e:\n                    attempt += 1\n                    last_exception = e\n                    if attempt < max_attempts:\n                        print(f\"[{func.__name__}] 失敗（試行 {attempt}/{max_attempts}）。\" \n                              f\"{delay}秒後に再試行...\")\n                        time.sleep(delay)\n                    else:\n                        print(f\"[{func.__name__}] {max_attempts}回の試行後も失敗\")\n            \n            raise last_exception\n        return wrapper\n    return decorator\n\n@retry_on_exception(max_attempts=3, delay=0.5, exceptions=(ConnectionError, TimeoutError))\ndef call_external_api(endpoint: str) -> dict:\n    \"\"\"外部APIを呼び出す\"\"\"\n    # 実装例：10%の確率で失敗\n    if random.random() < 0.1:\n        raise ConnectionError(f\"API接続失敗: {endpoint}\")\n    return {\"status\": \"success\", \"data\": []}\n\n# 使用例\ntry:\n    result = call_external_api(\"/api/users\")\n    print(result)\nexcept Exception as e:\n    print(f\"エラー: {e}\")

3. ログ記録デコレータ

関数の入出力をログに記録することで、デバッグやモニタリングが容易になります。本番環境でもよく使う重要なパターンです。

import functools\nimport logging\nfrom typing import Any, Callable\nimport json\n\nlogger = logging.getLogger(__name__)\n\ndef log_function_call(log_args: bool = True, log_result: bool = True) -> Callable:\n    \"\"\"関数の呼び出し、引数、戻り値をログに記録するデコレータ\"\"\"\n    def decorator(func: Callable) -> Callable:\n        @functools.wraps(func)\n        def wrapper(*args, **kwargs) -> Any:\n            func_name = func.__name__\n            \n            # 入力ログ\n            if log_args:\n                logger.info(\n                    f\"関数開始: {func_name}\",\n                    extra={\n                        \"function\": func_name,\n                        \"args\": str(args)[:200],  # 長すぎる場合はカット\n                        \"kwargs\": json.dumps(kwargs, default=str, ensure_ascii=False)[:200]\n                    }\n                )\n            \n            try:\n                result = func(*args, **kwargs)\n                \n                # 出力ログ\n                if log_result:\n                    logger.info(\n                        f\"関数終了: {func_name}\",\n                        extra={\n                            \"function\": func_name,\n                            \"result\": str(result)[:200]\n                        }\n                    )\n                \n                return result\n            except Exception as e:\n                logger.exception(\n                    f\"関数エラー: {func_name}\",\n                    extra={\n                        \"function\": func_name,\n                        \"exception\": str(e)\n                    }\n                )\n                raise\n        \n        return wrapper\n    return decorator\n\n@log_function_call(log_args=True, log_result=True)\ndef process_order(order_id: int, amount: float) -> dict:\n    \"\"\"注文を処理する\"\"\"\n    if amount <= 0:\n        raise ValueError(\"金額は正数である必要があります\")\n    return {\"order_id\": order_id, \"status\": \"completed\", \"amount\": amount}\n\n# 使用例\nresult = process_order(12345, 9999.99)

4. キャッシング機能付きデコレータ

同じ引数で関数が呼ばれた場合、前回の計算結果を再利用します。頻繁に同じクエリを実行するDB検索やAPI呼び出しで大きな効果があります。

import functools\nfrom typing import Any, Callable, Dict, Hashable\nimport time\n\ndef memoize_with_ttl(ttl_seconds: int = 60) -> Callable:\n    \"\"\"計算結果をキャッシュし、TTL付きで無効化するデコレータ\"\"\"\n    def decorator(func: Callable) -> Callable:\n        cache: Dict[Hashable, tuple] = {}  # (result, timestamp)\n        \n        @functools.wraps(func)\n        def wrapper(*args, **kwargs) -> Any:\n            # キャッシュキーを生成\n            cache_key = (args, tuple(sorted(kwargs.items())))\n            \n            # キャッシュを確認\n            if cache_key in cache:\n                result, cached_time = cache[cache_key]\n                elapsed = time.time() - cached_time\n                \n                if elapsed < ttl_seconds:\n                    print(f\"[{func.__name__}] キャッシュから取得（{elapsed:.1f}秒経過）\")\n                    return result\n                else:\n                    # TTL切れ\n                    del cache[cache_key]\n            \n            # 新規実行\n            result = func(*args, **kwargs)\n            cache[cache_key] = (result, time.time())\n            print(f\"[{func.__name__}] 新規実行してキャッシュに保存\")\n            return result\n        \n        # キャッシュをクリアするメソッドを追加\n        def clear_cache():\n            cache.clear()\n        \n        wrapper.clear_cache = clear_cache\n        return wrapper\n    \n    return decorator\n\n@memoize_with_ttl(ttl_seconds=5)\ndef get_user_info(user_id: int) -> dict:\n    \"\"\"ユーザー情報を取得（キャッシュあり）\"\"\"\n    time.sleep(1)  # 重い処理を想定\n    return {\"user_id\": user_id, \"name\": f\"User{user_id}\"}\n\n# 使用例\nprint(get_user_info(1))  # 新規実行\nprint(get_user_info(1))  # キャッシュから取得\ntime.sleep(6)\nprint(get_user_info(1))  # TTL切れなので新規実行\nget_user_info.clear_cache()  # 手動でキャッシュクリア

5. バリデーションデコレータ

関数の引数を自動検証し、不正な入力を事前に防ぎます。ビジネスロジックの前に入力チェックを一元化できます。

import functools\nfrom typing import Any, Callable, Dict, Optional\n\ndef validate_input(\n    schema: Dict[str, type],\n    required_keys: Optional[list] = None\n) -> Callable:\n    \"\"\"辞書形式の入力を検証するデコレータ\"\"\"\n    if required_keys is None:\n        required_keys = list(schema.keys())\n    \n    def decorator(func: Callable) -> Callable:\n        @functools.wraps(func)\n        def wrapper(*args, **kwargs) -> Any:\n            # 最初の引数が辞書として検証対象と仮定\n            if args and isinstance(args[0], dict):\n                data = args[0]\n                \n                # 必須キーのチェック\n                missing_keys = set(required_keys) - set(data.keys())\n                if missing_keys:\n                    raise ValueError(f\"必須キーが不足しています: {missing_keys}\")\n                \n                # 型チェック\n                for key, expected_type in schema.items():\n                    if key in data and not isinstance(data[key], expected_type):\n                        actual_type = type(data[key]).__name__\n                        expected_type_name = expected_type.__name__\n                        raise TypeError(\n                            f\"キー '{key}' の型が不正です。\"\n                            f\"期待: {expected_type_name}, 実際: {actual_type}\"\n                        )\n            \n            return func(*args, **kwargs)\n        return wrapper\n    return decorator\n\n@validate_input(\n    schema={\n        \"user_id\": int,\n        \"email\": str,\n        \"age\": int,\n        \"is_active\": bool\n    },\n    required_keys=[\"user_id\", \"email\"]\n)\ndef create_user(data: dict) -> dict:\n    \"\"\"ユーザーを作成する\"\"\"\n    return {\"status\": \"created\", \"data\": data}\n\n# 使用例\ntry:\n    # 成功例\n    result = create_user({\n        \"user_id\": 1,\n        \"email\": \"user@example.com\",\n        \"age\": 30,\n        \"is_active\": True\n    })\n    print(\"成功:\", result)\n    \n    # 失敗例：必須キー不足\n    create_user({\"user_id\": 2})  # EmailがないのでValueError\nexcept (ValueError, TypeError) as e:\n    print(f\"検証エラー: {e}\")

import functools\nimport time\n\ndef timer(func):\n    @functools.wraps(func)\n    def wrapper(*args, **kwargs):\n        start = time.time()\n        result = func(*args, **kwargs)\n        print(f\"実行時間: {time.time() - start:.3f}秒\")\n        return result\n    return wrapper\n\ndef logger_dec(func):\n    @functools.wraps(func)\n    def wrapper(*args, **kwargs):\n        print(f\"実行開始: {func.__name__}\")\n        result = func(*args, **kwargs)\n        print(f\"実行終了: {func.__name__}\")\n        return result\n    return wrapper\n\n# 複数のデコレータを適用（下から順に適用される）\n@timer\n@logger_dec\ndef complex_calculation(n: int) -> int:\n    \"\"\"複雑な計算を行う\"\"\"\n    time.sleep(0.5)\n    return n * n\n\nresult = complex_calculation(10)\n# 出力:\n# 実行開始: complex_calculation\n# 実行終了: complex_calculation\n# 実行時間: 0.502秒

クラスメソッド用デコレータ

クラスメソッドに対してもデコレータを適用できます。データベースアクセスオブジェクト（DAO）やAPIクライアントの実装で頻繁に使われます。

import functools\nfrom typing import Any, Callable\n\ndef requires_database_connection(func: Callable) -> Callable:\n    \"\"\"データベース接続をチェックするデコレータ\"\"\"\n    @functools.wraps(func)\n    def wrapper(self, *args, **kwargs) -> Any:\n        if not hasattr(self, 'db_connection') or self.db_connection is None:\n            raise RuntimeError(f\"データベース接続が確立されていません\")\n        return func(self, *args, **kwargs)\n    return wrapper\n\nclass UserRepository:\n    def __init__(self, db_connection=None):\n        self.db_connection = db_connection\n    \n    @requires_database_connection\n    def get_user(self, user_id: int) -> dict:\n        \"\"\"ユーザーを取得\"\"\"\n        # 実際のDB処理\n        return {\"user_id\": user_id, \"name\": \"太郎\"}\n    \n    @requires_database_connection\n    def create_user(self, name: str, email: str) -> dict:\n        \"\"\"ユーザーを作成\"\"\"\n        return {\"id\": 123, \"name\": name, \"email\": email}\n\n# 使用例\nrepo = UserRepository(db_connection=None)\ntry:\n    repo.get_user(1)  # RuntimeError: データベース接続が確立されていません\nexcept RuntimeError as e:\n    print(f\"エラー: {e}\")\n\n# 接続を確立\nrepo.db_connection = \"connected\"\nresult = repo.get_user(1)  # 成功\nprint(result)

import functools\n\n# 悪い例\ndef bad_decorator(func):\n    def wrapper(*args, **kwargs):\n        return func(*args, **kwargs)\n    return wrapper\n\n@bad_decorator\ndef my_func():\n    \"\"\"これは重要なドキュメント\"\"\"\n    pass\n\nprint(my_func.__name__)  # 'wrapper' （元の名前が失われている）\nprint(my_func.__doc__)   # None （ドキュメントが失われている）\n\n# 良い例\ndef good_decorator(func):\n    @functools.wraps(func)  # これが重要\n    def wrapper(*args, **kwargs):\n        return func(*args, **kwargs)\n    return wrapper\n\n@good_decorator\ndef my_func2():\n    \"\"\"これは重要なドキュメント\"\"\"\n    pass\n\nprint(my_func2.__name__)  # 'my_func2' （元の名前を保持）\nprint(my_func2.__doc__)   # \"これは重要なドキュメント\" （ドキュメントを保持）

2. 例外処理とリソース管理

デコレータで前処理と後処理を行う場合、例外が発生しても後処理が実行されるようにする必要があります。try-finallyやコンテキストマネージャを活用します。

import functools\nfrom contextlib import contextmanager\n\ndef with_resource_management(func):\n    \"\"\"リソース管理を行うデコレータ\"\"\"\n    @functools.wraps(func)\n    def wrapper(*args, **kwargs):\n        # リソース取得\n        resource = acquire_resource()\n        print(f\"リソース取得: {resource}\")\n        \n        try:\n            # 関数実行\n            result = func(*args, **kwargs)\n            return result\n        except Exception as e:\n            print(f\"エラー発生: {e}\")\n            raise\n        finally:\n            # 必ず実行（例外が発生してもリソースを解放）\n            release_resource(resource)\n            print(f\"リソース解放: {resource}\")\n    \n    return wrapper\n\ndef acquire_resource():\n    return \"DB_CONNECTION\"\n\ndef release_resource(resource):\n    pass\n\n@with_resource_management\ndef database_operation():\n    \"\"\"DB操作を実行\"\"\"\n    print(\"DB処理中...\")\n    # エラーが発生してもリソースは解放される\n    return \"success\"\n\nresult = database_operation()\n# 出力:\n# リソース取得: DB_CONNECTION\n# DB処理中...\n# リソース解放: DB_CONNECTION

3. パフォーマンスへの配慮

デコレータは全関数実行時に動作するため、過度なオーバーヘッドがないか注意が必要です。特にループ内で頻繁に呼ばれる関数にデコレータを適用する場合は、性能測定が重要です。

import functools\nimport time\n\ndef performance_critical_decorator(func):\n    \"\"\"最小限のオーバーヘッドで機能するデコレータ\"\"\"\n    @functools.wraps(func)\n    def wrapper(*args, **kwargs):\n        # 軽量な処理のみ\n        # ログ出力、重い計算は避ける\n        return func(*args, **kwargs)\n    return wrapper\n\n# パフォーマンス測定\nimport timeit\n\ndef regular_function(x):\n    return x * 2\n\n@performance_critical_decorator\ndef decorated_function(x):\n    return x * 2\n\n# 100万回の実行時間を計測\nregular_time = timeit.timeit(\n    lambda: regular_function(5),\n    number=1000000\n)\ndecorated_time = timeit.timeit(\n    lambda: decorated_function(5),\n    number=1000000\n)\n\nprint(f\"通常: {regular_time:.3f}秒\")\nprint(f\"デコレータ: {decorated_time:.3f}秒\")\nprint(f\"オーバーヘッド: {(decorated_time - regular_time) * 1000:.3f}ms\")

4. スタック可能性の確保

複数のデコレータを組み合わせる場合、引数の形式に統一性を持たせることが重要です。パラメータなしのデコレータとパラメータありのデコレータを混在させるときは特に注意が必要です。

import functools\nfrom typing import Any, Callable, Optional\n\n# パラメータなし\ndef simple_decorator(func: Callable) -> Callable:\n    @functools.wraps(func)\n    def wrapper(*args, **kwargs) -> Any:\n        print(\"Simple decorator\")\n        return func(*args, **kwargs)\n    return wrapper\n\n# パラメータあり\ndef parameterized_decorator(param: str = \"default\") -> Callable:\n    def decorator(func: Callable) -> Callable:\n        @functools.wraps(func)\n        def wrapper(*args, **kwargs) -> Any:\n            print(f\"Parameterized decorator: {param}\")\n            return func(*args, **kwargs)\n        return wrapper\n    return decorator\n\n# 統一的に組み合わせられるように\n@simple_decorator\n@parameterized_decorator(param=\"custom\")\ndef my_function():\n    print(\"Function execution\")\n\nmy_function()\n# 出力:\n# Simple decorator\n# Parameterized decorator: custom\n# Function execution

import functools\nimport time\nimport logging\nfrom typing import Any, Callable, Dict, Optional\nimport jwt\nfrom datetime import datetime\n\nlogger = logging.getLogger(__name__)\n\n# 1. 認証デコレータ\ndef require_auth(func: Callable) -> Callable:\n    \"\"\"APIの認証をチェック\"\"\"\n    @functools.wraps(func)\n    def wrapper(*args, request=None, **kwargs) -> Any:\n        if request is None or not hasattr(request, 'headers'):\n            raise PermissionError(\"リクエストが不正です\")\n        \n        token = request.headers.get('Authorization')\n        if not token:\n            raise PermissionError(\"トークンが見つかりません\")\n        \n        try:\n            # トークン検証（簡略版）\n            payload = jwt.decode(token.replace('Bearer ', ''), 'secret', algorithms=['HS256'])\n            request.user = payload\n        except jwt.InvalidTokenError:\n            raise PermissionError(\"無効なトークンです\")\n        \n        return func(*args, request=request, **kwargs)\n    \n    return wrapper\n\n# 2. レート制限デコレータ\ndef rate_limit(max_calls: int = 100, time_window: int = 60) -> Callable:\n    \"\"\"時間ウィンドウ内での呼び出し回数を制限\"\"\"\n    def decorator(func: Callable) -> Callable:\n        call_times = {}\n        \n        @functools.wraps(func)\n        def wrapper(*args, user_id: Optional[str] = None, **kwargs) -> Any:\n            if user_id is None:\n                user_id = \"anonymous\"\n            \n            now = time.time()\n            if user_id not in call_times:\n                call_times[user_id] = []\n            \n            # 時間ウィンドウ外の呼び出しを削除\n            call_times[user_id] = [\n                call_time for call_time in call_times[user_id]\n                if now - call_time < time_window\n            ]\n            \n            if len(call_times[user_id]) >= max_calls:\n                raise RuntimeError(\n                    f\"レート制限超過: {max_calls}回/{time_window}秒\"\n                )\n            \n            call_times[user_id].append(now)\n            return func(*args, user_id=user_id, **kwargs)\n        \n        return wrapper\n    return decorator\n\n# 3. ログ+タイミング+エラー処理\ndef api_handler(\n    operation_name: str,\n    log_request_body: bool = False\n) -> Callable:\n    \"\"\"API処理の標準的なハンドリング\"\"\"\n    def decorator(func: Callable) -> Callable:\n        @functools.wraps(func)\n        def wrapper(*args, request=None, **kwargs) -> Any:\n            request_id = getattr(request, 'id', 'unknown')\n            start_time = time.time()\n            \n            logger.info(\n                f\"[{request_id}] API開始: {operation_name}\",\n                extra={\n                    \"operation\": operation_name,\n                    \"timestamp\": datetime.now().isoformat()\n                }\n            )\n            \n            try:\n                result = func(*args, request=request, **kwargs)\n                \n                elapsed = time.time() - start_time\n                logger.info(\n                    f\"[{request_id}] API成功: {operation_name}\",\n                    extra={\n                        \"operation\": operation_name,\n                        \"elapsed_ms\": elapsed * 1000,\n                        \"status\": \"success\"\n                    }\n                )\n                \n                return result\n            \n            except Exception as e:\n                elapsed = time.time() - start_time\n                logger.error(\n                    f\"[{request_id}] API失敗: {operation_name}\",\n                    extra={\n                        \"operation\": operation_name,\n                        \"elapsed_ms\": elapsed * 1000,\n                        \"error\": str(e),\n                        \"error_type\