문제 설명 − Python의 boto3 라이브러리를 사용하여 웨이터 기능을 사용하여 버킷에 키가 존재하지 않는지 확인합니다. 예를 들어, 웨이터를 사용하여 키 test1.zip이 Bucket_1에 존재하지 않는지 확인하십시오.
이 문제를 해결하기 위한 접근 방식/알고리즘
1단계 − boto3 및 botocore 예외를 가져와 예외를 처리합니다.
2단계 − bucket_name과 key는 함수의 두 매개변수입니다.
3단계 − boto3 라이브러리를 사용하여 AWS 세션을 생성합니다.
4단계 − S3용 AWS 클라이언트를 생성합니다.
5단계 − 이제 get_waiter 함수를 사용하여 object_not_exists에 대한 대기 개체를 만듭니다.
6단계 − 이제 wait 객체를 사용하여 지정된 버킷에 키가 존재하지 않는지 확인합니다. 기본적으로 성공적인 상태에 도달할 때까지 5초마다 체크인합니다. 20번의 검사 실패 후 오류가 반환됩니다. 그러나 사용자가 폴링 시간과 최대 시도 횟수를 정의할 수 있습니다.
7단계 − None을 반환합니다.
8단계 − 버킷을 확인하는 동안 문제가 발생한 경우 일반 예외를 처리합니다.
예시
키가 버킷에 존재하는지 여부를 확인하기 위해 waiter를 사용하려면 다음 코드를 사용하십시오 -
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의 경우 출력은 print 문이고 None입니다. 응답은 아무 것도 반환하지 않으므로 None을 출력합니다.
Bucket_1/testfolder/test.zip의 경우 이 개체가 존재하므로 출력은 예외입니다.
예외적으로 최대 시도 횟수를 초과한 것으로 읽을 수 있습니다.