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