boto3でエラーハンドリングする方法をまとめる | AWS SDK for Python

boto3でエラーハンドリングする方法を整理します。

はじめに
boto3でエラーハンドリングする
おわりに
参考

はじめに

こんにちは、@bioerrorlogです。

boto3で発生するエラーをcatchしようとしたときに、少しハマったことがありました。

例えば以下のコードを実行すると、examplebucketという名前のS3バケットが既に存在しているため、BucketAlreadyExistsエラーが吐かれます。

import boto3


client = boto3.client('s3')
client.create_bucket(
    Bucket='examplebucket',
    CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1',},
)

botocore.errorfactory.BucketAlreadyExists: An error occurred (BucketAlreadyExists) when calling the CreateBucket operation: The requested bucket name is not available.
The bucket namespace is shared by all users of the system. Please select a different name and try again.

では、このエラーをcatchするにはどうすればいいでしょうか。

エラーメッセージに従って愚直にbotocore.errorfactory.BucketAlreadyExistsをcatchする例外処理を書いてしまうと、上手く動きません。

import boto3
import botocore


client = boto3.client('s3')
try:
    client.create_bucket(
        Bucket='examplebucket',
        CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1',},
    )
except botocore.errorfactory.BucketAlreadyExists as e:
    print(e)

AttributeError: module 'botocore.errorfactory' has no attribute 'BucketAlreadyExists'

今回は、このようなケースでboto3エラーをハンドリングする方法をまとめます。

boto3でエラーハンドリングする

boto3でエラーハンドリングするやり方は、大きく分けて２つあります。

client.exceptions
botocore.exceptions

client.exceptions

client.exceptionsを使えば、とてもシンプルに例外処理を書くことが出来ます。宣言したboto3 clientから、client.exceptions.<エラー名>でcatchします。

import boto3


client = boto3.client('s3')
try:
    client.create_bucket(
        Bucket='examplebucket',
        CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1',},
    )
except client.exceptions.BucketAlreadyExists as e:
    print(e.response['Error'])

なお、boto3 clientではなくresourceを宣言している場合はresource.meta.client.exceptions.<エラー名>でcatchできます。

import boto3


resource = boto3.resource('s3')
try:
    resource.create_bucket(
        Bucket='examplebucket',
        CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1',},
    )
except resource.meta.client.exceptions.BucketAlreadyExists as e:
    print(e.response['Error'])

ただし、全てのエラーがこのやり方でcatch出来るわけではない、という点に注意が必要です。対応できるエラーは、boto3ドキュメントの各APIごとに記載があります。

一方、次に紹介するbotocore.exceptionsを使ったやり方では、全てのエラーをcatchできます。

botocore.exceptions

botocore.exceptionsを使う場合は少し冗長な書き方になってしまいますが、全ての例外を処理できるメリットがあります。

import botocore
import boto3


client = boto3.client('s3')
try:
    client.create_bucket(
        Bucket='examplebucket',
        CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1',},
    )
except botocore.exceptions.ClientError as e:
    if e.response['Error']['Code'] == 'BucketAlreadyExists':
        print(e.response['Error'])
    else:
        raise e

botocore.exceptions.ClientErrorでエラーをcatchしたのち、エラーのresponse['Error']['Code']から対応を振り分けます。

コード量が少し多くなってしまうのでなるべくclient.exceptionsを使ったやり方で書きたいところですが、やむを得ない場合はこちらで実装しましょう。

コード例

では、いくつかboto3で例外処理をするコード例を書いてみます。 xxxAlreadyExists系エラーやNoSuchEntitiy系のエラーはcatchしたいケースが多いので、これら中心にメモします。

S3 create_bucketにおけるBucketAlreadyExistsエラーのハンドリング例

import boto3


client = boto3.client('s3')
try:
    client.create_bucket(
        Bucket='examplebucket',
        CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1',},
    )
except client.exceptions.BucketAlreadyExists as e:
    print(f'S3 bucket already exists: {e.response["Error"]["BucketName"]}')

S3 get_objectにおけるNoSuchKeyエラーのハンドリング例

import boto3


client = boto3.client('s3')
try:
    client.get_object(
        Bucket='target-bucket-001',
        Key='no-such/ocject-key',
    )
except client.exceptions.NoSuchKey as e:
    print(f'No such key: {e.response["Error"]["Key"]}')

IAM get_roleにおけるNoSuchEntityExceptionエラーのハンドリング例

import boto3


client = boto3.client('iam')
try:
    client.get_role(
        RoleName='NoSuchRoleName'
    )
except client.exceptions.NoSuchEntityException as e:
    print(e.response['Error']['Message'])

Glue create_databaseにおけるAlreadyExistsExceptionエラーのハンドリング例

import boto3


client = boto3.client('glue')
try:
    client.create_database(
        DatabaseInput={
            'Name': 'test_db'
        }
    )
except client.exceptions.AlreadyExistsException as e:
    print(e.response['Error']['Message'])