如何使用 Waiters 检查 S3 bucket 是否存在，使用 Boto3 和 AWS Client?
boto3pythonserver side programmingprogramming更新于 2024/1/10 23:06:00
在本文中，我们将了解如何使用 Boto3 库和 Waiter 功能来验证 S3 bucket 是否存在。例如，使用 waiters 检查 S3 中是否存在 Bucket_1
解决此问题的方法/算法
步骤 1 − 导入 boto3 和 botocore 异常来处理异常。
步骤 2 − 使用 bucket_name 作为函数中的参数。
步骤 3 − 使用 boto3 库创建 AWS 会话。
步骤 4 − 为 S3 创建 AWS 客户端。
步骤 5 −现在使用 get_waiter 函数为 bucket_exists 创建等待对象。
步骤 6 − 现在，使用等待对象验证 bucket 是否存在。默认情况下，它每 5 秒检查一次，直到达到成功状态。20 次检查失败后将返回错误。但是，用户可以定义轮询时间和最大尝试次数。
步骤 7 − 它返回 None。
步骤 8 −如果检查存储桶时出现错误，则处理通用异常。
示例
使用以下代码使用 waiter 检查 bucket_exists 是否 −
Use the following code to use waiter to check whether bucket_exists or not −
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，输出是打印语句和 None。由于响应未返回任何内容，因此它会打印 None。
对于 Bucket_2，输出是异常，因为此存储桶不存在。
在异常中，可以读取最大尝试次数超出。
