AWS Python(Boto3)の例外ハンドリングとテスト時のraise方法

Lambdaなど、AWSリソースを使用した開発をしているとboto3のエラーハンドリングをする機会が必ず出てきます。
また、Pytestなどテスト自動化を導入している際にraiseさせたい場面もよくあります。
そこで、エラーハンドリングをする際の手順を簡単にまとめました。

ハンドリング方法

• DynamoDBへレコード追加
• ID重複時に重複している旨を出力する

という想定で行っていきます。

エラーハンドリング

import boto3
from botocore.exceptions import ClientError
# boto3のエラーを司るClientErrorをimportしておく

dynamodb_client = boto3.client('dynamodb')


def main(table_name, id_):
    param = {
        "TableName": table_name,
        "Item": {
            "id": {"S": id_}
        },
        "Expected": {
            "id": {
                "Exists": False
            }
        }
    }

    try:
        dynamodb_client.put_item(**param)
    except ClientError as e:
        # ClientErrorをキャッチするようにする
        # エラー内容からCodeを抜き取り比較する。重複時にはConditionalCheckFailedExceptionが返ってくる。
        if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
            print("IDが重複しています。")
            return

        # 重複以外はエラー
        raise e


if __name__ == "__main__":
    main('hogehoge', 'fugafuga')

一回目は正常に動きますが、2回目はIDが重複しています。とメッセージが出ます。

ClientErrorのエラー内容を確認

VSCodeのデバッガを使用して中身を確認してみました。

御覧の通り、response内にErrorのキーがあり、さらにCodeのキーがあるのが確認できると思います。
エラーコード自体はAWSの公式リファレンスに載っていますので、事前に確認しておくのが良いと思います。
AWS Dynamodb | エラー処理

raise方法

ここまでは他にもたくさんqiitaの記事がありますが、raiseさせる方法を記述している記事はなかなか無かったのでメモがてら。

結論から言うとraiseするClientErrorに引数を渡してあげます。

try内でraiseさせる

# 先ほどのコードのtry,except部分のみ抜粋です
    try:
        # "Error"キーと"Code"キーの辞書型を用意
        # "Code"にはハンドリングさせたいエラーコードを入力
        error_response = {
            "Error": {
                "Code": "ConditionalCheckFailedException"
            }
        }
        # 文字列です。関数名などを格納する？
        operation_name = "put_item"
        raise ClientError(error_response, operation_name)
        dynamodb_client.put_item(**param)
    except ClientError as e:
        # ClientErrorをキャッチするようにする
        # エラー内容からCodeを抜き取り比較する。重複時にはConditionalCheckFailedExceptionが返ってくる。
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            print("IDが重複しています。")
            return

これでraiseできます。

raise ClientErrorのエラー内容を確認

若干メンバが足りないようですが、動作確認などには十分ですね。

ClientErrorの実態

ClientErrorはどのような構成になっているのか見てみました。

ClientError

class ClientError(Exception):
    MSG_TEMPLATE = (
        'An error occurred ({error_code}) when calling the {operation_name} '
        'operation{retry_info}: {error_message}')

    def __init__(self, error_response, operation_name):
        retry_info = self._get_retry_info(error_response)
        error = error_response.get('Error', {})
        msg = self.MSG_TEMPLATE.format(
            error_code=error.get('Code', 'Unknown'),
            error_message=error.get('Message', 'Unknown'),
            operation_name=operation_name,
            retry_info=retry_info,
        )
        super(ClientError, self).__init__(msg)
        self.response = error_response
        self.operation_name = operation_name

(続く)...

こちらがClientErrorの中身です。__init__で2つの引数を受け取ることができるのが読み取れます。
第1引数のerror_responseは辞書型であり、Errorがありその下にCodeが存在していればいいみたいです。