Boto3のWaiter（ウェイター）機能を使ってS3バケットの存在を確認する方法

この記事では、Boto3ライブラリのWaiter（ウェイター）機能を使って、指定したS3バケットが存在するかどうかを確認する方法を解説します。例として、「Bucket_1」というバケットがS3上に存在するかどうかをWaiterでチェックしてみましょう。

問題を解決するためのアプローチ／アルゴリズム

ステップ1 − 例外処理のために、boto3とbotocoreの例外クラスをインポートします。

ステップ2 − 関数のパラメータとしてbucket_nameを受け取ります。

ステップ3 − boto3ライブラリを使用してAWSセッションを作成します。

ステップ4 − S3用のAWSクライアントを作成します。

ステップ5 − get_waiter関数を使って、bucket_exists用のウェイトオブジェクトを作成します。

ステップ6 − ウェイトオブジェクトを使って、バケットが存在するかどうかを検証します。デフォルトでは5秒ごとにチェックが行われ、成功状態に達するまで繰り返されます。20回チェックが失敗するとエラーが返されます。なお、ポーリング間隔（Delay）と最大試行回数（MaxAttempts）はユーザーが自由に設定可能です。

ステップ7 − 正常終了時の戻り値はNoneです。

ステップ8 − バケットの確認中に何らかの問題が発生した場合に備え、汎用的な例外処理を実装します。

コード例

以下のコードを使用すると、Waiterによってバケットの存在有無をチェックできます。

import boto3
from botocore.exceptions import ClientError

def use_waiters_check_bucket_exists(bucket_name):
    session = boto3.session.Session(profile_name='saml')
    s3_client = session.client('s3')
    try:
        waiter = s3_client.get_waiter('bucket_exists')
        waiter.wait(Bucket=bucket_name,
                    WaiterConfig={
                        'Delay': 2, 'MaxAttempts': 5})
        print('Bucket exists: ' + bucket_name)
    except ClientError as e:
        raise Exception( "boto3 client error in use_waiters_check_bucket_exists: " + e.__str__())
    except Exception as e:
        raise Exception( "Unexpected error in use_waiters_check_bucket_exists: " + e.__str__())

print(use_waiters_check_bucket_exists("Bucket_1"))
print(use_waiters_check_bucket_exists("Bucket_2"))

実行結果

Bucket exists: Bucket_1
None

botocore.exceptions.WaiterError: Waiter BucketExists failed: Max
attempts exceeded
"Unexpected error in use_waiters_check_bucket_exists: " + e.__str__())
Exception: Unexpected error in use_waiters_check_bucket_exists: Waiter
BucketExists failed: Max attempts exceed

Bucket_1の場合、出力はprint文のメッセージとNoneになります。wait()のレスポンスは何も返さないため、Noneが表示されるのです。

一方、Bucket_2の場合は、そのバケットが存在しないため例外が発生します。

例外メッセージには「Max attempts exceeded（最大試行回数を超過）」という内容が含まれており、設定した試行回数内にバケットの存在が確認できなかったことを示しています。