如何使用 Wait 功能来检查 S3 存储桶中是否存在某个键,使用 Boto3 和 AWS 客户端?


当用户希望使用 wait 功能来验证程序代码中存储桶中是否存在某个键时。

问题陈述 - 使用 Python 中的 boto3 库来检查存储桶中是否存在某个键,使用 waiters 功能。例如,使用 waiters 检查 Bucket_1 中是否存在键 test.zip。

解决此问题的方法/算法

步骤 1 - 导入 boto3 和 botocore 异常以处理异常。

步骤 2 - bucket_name 和 key 是函数中的两个参数。

步骤 3 - 使用 boto3 库创建 AWS 会话。

步骤 4 - 为 S3 创建 AWS 客户端。

步骤 5 - 现在使用 get_waiter 函数为 object_exists 创建 wait 对象。

步骤 6 - 现在,使用 wait 对象来验证给定存储桶中是否存在键。默认情况下,它每 5 秒检查一次,直到达到成功状态。在 20 次失败检查后返回错误。但是,用户可以定义轮询时间和最大尝试次数。

步骤 7 - 它返回 None。

步骤 8 - 如果在检查存储桶时出现问题,则处理通用异常。

示例

使用以下代码使用 waiter 来检查存储桶中是否存在某个键 -

import boto3
from botocore.exceptions import ClientError

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

print(use_waiters_check_object_exists("Bucket_1","testfolder/test.zip"))
print(use_waiters_check_object_exists("Bucket_1","testfolder/test1.zip")
)

输出

Object exists: Bucket_1/testfolder/test.zip
None

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

对于 Bucket_1/testfolder/test.zip,输出是 print 语句和 None。由于响应没有返回任何内容,因此它打印 None。

对于 Bucket_1/testfolder/test1.zip,输出是异常,因为此对象不存在。

在异常中,可以读取“最大尝试次数已超过”。

更新于: 2021年3月22日

2K+ 次查看

开启你的 职业生涯

通过完成课程获得认证

开始学习
广告

© . All rights reserved.