如何使用 Boto3 和 AWS 客户端的 waiter 功能来检查 bucket 是否不存在?
问题陈述 − 使用 Python 中的 boto3 库,使用 waiter 功能检查 bucket 是否不存在。例如,使用 waiter 检查 S3 中 Bucket_2 是否不存在。
解决此问题的方法/算法
步骤 1 − 导入 boto3 和 botocore 异常以处理异常。
步骤 2 − 使用 bucket_name 作为函数中的参数。
步骤 3 − 使用 boto3 库创建 AWS 会话。
步骤 4 − 为 S3 创建 AWS 客户端。
步骤 5 − 现在使用 get_waiter 函数创建 bucket_not_exists 的等待对象。
步骤 6 − 现在,使用 wait 对象验证 bucket 是否不存在。默认情况下,它每 5 秒检查一次,直到达到成功状态。这里成功状态意味着 bucket 不应该存在于 S3 中。在 20 次失败检查后返回错误。但是,用户可以定义轮询时间和最大尝试次数。
步骤 7 − 它返回 None。
步骤 8 − 如果在检查 bucket 期间出现任何错误,请处理通用异常。
示例
使用以下代码使用 waiter 检查 bucket 是否不存在 −
import boto3
from botocore.exceptions import ClientError
def use_waiters_check_bucket_not_exists(bucket_name):
session = boto3.session.Session()
s3_client = session.client('s3')
try:
waiter = s3_client.get_waiter('bucket_not_exists')
waiter.wait(Bucket=bucket_name,
WaiterConfig={
'Delay': 2, 'MaxAttempts': 5})
print('Bucket not exist: ' + bucket_name)
except ClientError as e:
raise Exception( "boto3 client error in use_waiters_check_bucket_not_exists: " + e.__str__())
except Exception as e:
raise Exception( "Unexpected error in use_waiters_check_bucket_not_exists: " + e.__str__())
print(use_waiters_check_bucket_not_exists("Bucket_2"))
print(use_waiters_check_bucket_not_exists("Bucket_1"))输出
Bucket not exist: Bucket_2 None botocore.exceptions.WaiterError: Waiter BucketNotExists failed: Max attempts exceeded "Unexpected error in use_waiters_check_bucket_not_exists: " + e.__str__()) Exception: Unexpected error in use_waiters_check_bucket_not_exists: Waiter BucketNotExists failed: Max attempts exceed
对于 Bucket_2,输出是 print 语句和 None。由于响应没有返回任何内容,因此它打印 None。
对于 Bucket_1,输出是一个异常,因为即使在最大尝试次数检查后,此 bucket 仍然存在。
在异常中,可以读取最大尝试次数已超过。
广告
数据结构
网络
关系数据库管理系统
操作系统
Java
iOS
HTML
CSS
Android
Python
C 编程
C++
C#
MongoDB
MySQL
Javascript
PHP