이 기사에서는 Boto3 라이브러리와 Waiter 기능을 사용하여 S3 버킷의 존재 여부를 확인하는 방법을 살펴봅니다. 예를 들어 웨이터를 사용하여 Bucket_1이 S3에 있는지 확인합니다.
이 문제를 해결하기 위한 접근 방식/알고리즘
1단계 − boto3 및 botocore 예외를 가져와 예외를 처리합니다.
2단계 − 함수의 매개변수로 bucket_name을 사용합니다.
3단계 − boto3 라이브러리를 사용하여 AWS 세션을 생성합니다.
4단계 − S3용 AWS 클라이언트를 생성합니다.
5단계 − 이제 bucket_exists에 대한 대기 개체를 만듭니다. get_waiter 사용 기능.
6단계 − 이제 wait 객체를 사용하여 버킷의 존재 여부를 확인합니다. 기본적으로 성공적인 상태에 도달할 때까지 5초마다 체크인합니다. 20번의 검사 실패 후 오류가 반환됩니다. 그러나 사용자가 폴링 시간과 최대 시도 횟수를 정의할 수 있습니다.
7단계 − None을 반환합니다.
8단계 − 버킷을 확인하는 동안 문제가 발생한 경우 일반 예외를 처리합니다.
예시
다음 코드를 사용하여 웨이터를 사용하여 버킷이 존재하는지 여부를 확인하십시오 -
botocore.exceptions에서 boto3 가져오기 import ClientErrordef use_waiters_check_bucket_exists(bucket_name):session =boto3.session.Session(profile_name='saml') s3_client =session.client('s3') 시도:waiter =s3_client.get_waiter('bucket_exists ) waiter.wait(Bucket=bucket_name, WaiterConfig={ 'Delay':2, 'MaxAttempts':5}) print('버킷이 존재함:' + bucket_name) 예외로 ClientError as e:raise Exception( "boto3 client error in use_waiters_check_bucket_exists:" + e.__str__()) 예외 e:raise Exception( "use_waiters_check_bucket_exists에서 예기치 않은 오류 발생:" + e.__str__())print(use_waiters_check_bucket_exists("Bucket_1"))print(use_waiters_check_bucket_exists("Bucket_2") 출력
버킷이 존재함:Bucket_1Nonebotocore.exceptions.WaiterError:Waiter BucketExists 실패:Maxattempts added"Unexpected error in use_waiters_check_bucket_exists:" + e.__str__())Exception:use_waiters_check_bucket_exists의 예기치 않은 오류:WaiterBucketExists 초과 시도
Bucket_1의 경우 출력은 print 문 및 없음입니다. 응답은 아무 것도 반환하지 않으므로 None을 출력합니다.
Bucket_2의 경우 이 버킷이 존재하지 않으므로 출력은 예외입니다.
예외적으로 최대 시도 횟수를 초과한 것으로 읽을 수 있습니다.