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')

1번째는 정상적으로 움직입니다만, 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__ 에서 두 개의 인수를 받을 수 있는 것을 읽을 수 있습니다.
첫 번째 인수 error_response는 사전 형식이며 오류가 있고 그 아래에 코드가 있으면 좋을 것 같습니다.