[Python]デコレータを使ってboto3のエラーハンドリングを共通化してみた

鈴木純

2025.02.06

ClientErrorのハンドリングを共通化したい

boto3 を使った開発していると、実行する API ごとにエラーハンドリングするコードが増えてきました。

例えば以下のような実行したリージョンやアカウント ID の情報を合わせて出力したいケースです。

vpc_manager.py

import boto3
from aws_lambda_powertools.utilities.typing import LambdaContext
from mypy_boto3_ec2.type_defs import DescribeVpcsResultTypeDef

class VPCManager:

    def __init__(self, region: str) -> None:
        self.client = boto3.client("ec2", region_name=region)
        self.account_id = boto3.client("sts").get_caller_identity().get("Account")
        self.region = region

    def create_default_vpc(self) -> None:
        try:
            self.client.create_default_vpc()
        except ClientError as e:
            logger.error(f"[{self.account_id}] ({self.region}) Failed create_default_vpc")

    def describe_vpcs(self) -> DescribeVpcsResultTypeDef:
        try:
            self.client.describe_vpcs()
        except ClientError as e:
            logger.error(f"[{self.account_id}] ({self.region}) Failed describe_vpcs")

def handler(event: dict, context: LambdaContext) -> None:
    vpc_manager = VPCManager(region="ap-northeast-1")
    vpc_manager.create_default_vpc()

失敗した場合を想定して、全てに try-except を書いてリージョン名などを出力しています。

これは冗長的だなと思い、簡単に ClientError 時のエラー出力を共通化できるようデコレータを試してみました。

やってみる

必須ではないですが、ログ出力は見やすいようにAWS Lambda Powertoolsを使用しています。まだ使ったことがない方は以下をご参照ください。

デコレータについては以下の記事で勉強させてもらいました。簡単に言うと、関数を受け取ってその機能を拡張した関数を返す機能です。です。

デコレータの作成

実装したデコレータは以下です。

前提として、呼び出し元のインスタンスが account_id と region を持つようなケースを想定しています。

decorators.py

from typing import Any, Callable

from aws_lambda_powertools import Logger
from botocore.exceptions import ClientError

logger = Logger()

def log_client_exception(func: Any) -> Callable:
    """
    共通のエラーハンドリングデコレータ。
    ClientErrorが発生した際、呼び出し元のインスタンス（self）がaccount_idおよびregion属性を持つ場合、
    詳細な情報をログに出力します。
    """

    def wrapper(*args: Any, **kwargs: Any) -> Any:
        try:
            return func(*args, **kwargs)
        except ClientError as e:
            if args and hasattr(args[0], "account_id") and hasattr(args[0], "region"):
                instance = args[0]
                logger.error(
                    f"[{instance.account_id}] ({instance.region}) Error in {func.__name__}: {e}"
                )
            else:
                logger.error(f"Error in {func.__name__}: {e}")
            raise

    return wrapper

実行された関数でClientErrorが発生した場合、インスタンス内のアカウント ID とリージョン・関数名を含めたエラーログを出力するようにしました。

この関数を共通関数用のフォルダ等に配置し、インポートできる状態にします。

デコレータを使った共通化ログを確認する

先ほどのコードをデコレータを使ったものに修正したのがこちらです。

vpc_manager.py

import os

import boto3
from aws_lambda_powertools.utilities.typing import LambdaContext
+from decorators import log_client_exception

class VPCManager:

    def __init__(self, region: str) -> None:
        self.client = boto3.client("ec2", region_name=region)
        self.account_id = boto3.client("sts").get_caller_identity().get("Account")
        self.region = region

+   @log_client_exception
    def create_default_vpc(self) -> None:
        self.client.create_default_vpc()

+   @log_client_exception
    def describe_vpcs(self) -> dict:
        return self.client.describe_vpcs()

def handler(event: dict, context: LambdaContext) -> None:
    vpc_manager = VPCManager(region="ap-northeast-1")
    vpc_manager.create_default_vpc()

インポート文を追加して、対象の関数に先ほど作成したデコレータ@log_client_exceptionを追加しています。
try-except の記述がなくなったことで、create_default_vpc や describe_vpcs が非常にシンプルになりました。

Lambda 上で実行してClientErrorを確認します。すでにデフォルト VPC がある環境だったため、以下のエラーとなりました。

{
  "errorMessage": "An error occurred (DefaultVpcAlreadyExists) when calling the CreateDefaultVpc operation: A Default VPC already exists for this account in this region.",
  "errorType": "ClientError",
  "requestId": "f5e15ea7-4608-44e8-b0c5-d5767135610d",
  "stackTrace": [
    "  File \"/var/task/index.py\", line 23, in handler\n    vpc_manager.create_default_vpc()\n",
    "  File \"/opt/python/decorators.py\", line 18, in wrapper\n    return func(*args, **kwargs)\n",
    "  File \"/var/task/index.py\", line 18, in create_default_vpc\n    self.client.create_default_vpc()\n",
    "  File \"/var/lang/lib/python3.12/site-packages/botocore/client.py\", line 565, in _api_call\n    return self._make_api_call(operation_name, kwargs)\n",
    "  File \"/var/lang/lib/python3.12/site-packages/botocore/client.py\", line 1021, in _make_api_call\n    raise error_class(parsed_response, operation_name)\n"
  ]
}

CloudWatch Logs でデコレータによるエラーログを確認すると、以下の出力が出ていました。messageの欄にエラーが発生したアカウント ID、リージョン、関数名、発生したエラーメッセージが出ていますね。

{
    "level": "ERROR",
    "location": "wrapper:22",
    "message": "[111111111111] (ap-northeast-1) Error in create_default_vpc: An error occurred (DefaultVpcAlreadyExists) when calling the CreateDefaultVpc operation: A Default VPC already exists for this account in this region.",
    "timestamp": "2025-02-06 02:16:29,889+0000",
    "service": "service_undefined",
    "xray_trace_id": "1-67a41b78-7dec46c65d6d305f01760c42"
}

特定のエラーは個別に対応したい

特定のエラーのみ別の処理をしたいケースありますよね。その場合は普通通りに try-except で記述すれば問題ありません。
先ほどのDefaultVpcAlreadyExistsをスキップさせたい場合は、以下のような記述になります。

vpc_manager

class VPCManager:
    os.environ["POWERTOOLS_SERVICE_NAME"] = "VPCManager"

    def __init__(self, region: str) -> None:
        self.client = boto3.client("ec2", region_name=region)
        self.account_id = boto3.client("sts").get_caller_identity().get("Account")
        self.region = region

    @log_client_exception
    def create_default_vpc(self) -> None:
        try:
            self.client.create_default_vpc()
        except ClientError as e:
            if e.response.get("Error", {}).get("Code") == "DefaultVpcAlreadyExists":
                logger.info(
                    f"[{self.account_id}] ({self.region}) Default VPC already exists, skipping creation"
                )
                return
            raise

実行したログ結果はこちらです。Lambda は成功しており、messageに先ほど記述した INFO のメッセージが出ています。

{
    "level": "INFO",
    "location": "create_default_vpc:27",
    "message": "[111111111111] (ap-northeast-1) Default VPC already exists, skipping creation",
    "timestamp": "2025-02-06 04:33:31,849+0000",
    "service": "service_undefined",
    "xray_trace_id": "1-67a43b95-3a812adf060dab47306ca583"
}