如何在 Boto3 中使用 Wait 功能验证 S3 桶中是否存在某个键?
问题陈述 − 使用 Python 中的 boto3 库,利用等待器功能检查桶中是否存在某个键。例如,使用等待器检查 Bucket_1 中是否存在名为 test1.zip 的键。
解决此问题的方法/算法
步骤 1 − 导入 boto3 和 botocore 异常以处理异常。
步骤 2 − bucket_name 和 key 是函数中的两个参数。
步骤 3 − 使用 boto3 库创建一个 AWS 会话。
步骤 4 − 为 S3 创建一个 AWS 客户端。
步骤 5 − 现在使用 get_waiter 函数为 object_not_exists 创建等待对象。
步骤 6 − 现在,使用等待对象验证给定桶中是否存在键。默认情况下,它每 5 秒检查一次,直到达到成功状态。在 20 次失败检查后返回错误。但是,用户可以定义轮询时间和最大尝试次数。
步骤 7 − 它返回 None。
步骤 8 − 如果在检查桶时出现错误,则处理通用异常。
示例
使用以下代码使用等待器检查桶中是否存在键:
import boto3 from botocore.exceptions import ClientError def use_waiters_check_object_not_exists(bucket_name, key_name): session = boto3.session.Session() s3_client = session.client('s3') try: waiter = s3_client.get_waiter('object_not_exists') waiter.wait(Bucket=bucket_name, Key = key_name, WaiterConfig={ 'Delay': 2, 'MaxAttempts': 5}) print('Object does not exist: ' + bucket_name +'/'+key_name) except ClientError as e: raise Exception( "boto3 client error in use_waiters_check_object_not_exists: " + e.__str__()) except Exception as e: raise Exception( "Unexpected error in use_waiters_check_object_not_exists: " + e.__str__()) print(use_waiters_check_object_exists("Bucket_1","testfolder/test1.zip")) print(use_waiters_check_object_exists("Bucket_1","testfolder/test.zip"))
输出
Object does not exist: Bucket_1/testfolder/test1.zip None botocore.exceptions.WaiterError: Waiter ObjectNotExists failed: Max attempts exceeded "Unexpected error in use_waiters_check_object_not_exists: " + e.__str__()) Exception: Unexpected error in use_waiters_check_object_not_exists: Waiter ObjectNotExists failed: Max attempts exceed
对于 Bucket_1/testfolder/test1.zip,输出是打印语句和 None。由于响应没有返回任何内容,因此它打印 None。
对于 Bucket_1/testfolder/test.zip,由于此对象存在,因此输出是一个异常。
在异常中,可以读取“最大尝试次数已超过”。
广告