如何使用等待功能检查 S3 存储桶中的密钥是否存在,使用 Boto3 和 AWS 客户端?
当用户想要使用等待功能来验证存储桶中的密钥是否存在于编程代码中时。
问题陈述-使用Python中的boto3库,使用服务员功能检查存储桶中是否存在密钥。例如,使用test.zipwaiters检查Bucket_1中是否存在某个键。
解决这个问题的方法/算法
步骤1-导入boto3和botocore异常以处理异常。
Step2-bucket_name和key是函数中的两个参数。
步骤3-使用boto3库创建AWS会话。
第4步-为S3创建一个AWS客户端。
第5步-现在使用get_waiter函数为object_exists创建等待对象
第6步-现在,使用等待 对象来验证给定存储桶中是否存在密钥。默认情况下,它每5秒检查一次,直到达到成功状态。检查失败20次后返回错误。但是,用户可以定义轮询时间和最大尝试次数。
步骤7-它返回None。
第8步-如果在检查存储桶时出现问题,则处理通用异常。
示例
使用以下代码使用waiter 检查存储桶中是否存在密钥-
import boto3 frombotocore.exceptionsimport 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,输出是打印语句和无。由于响应不返回任何内容,因此它打印None。
对于Bucket_1/testfolder/test1.zip,输出是一个例外,因为该对象不存在。
作为例外,可以读取最大尝试次数超过。