Boto3 Waiter를 활용해 S3 버킷 존재 여부 확인하는 방법

이 글에서는 Boto3 라이브러리와 Waiter 기능을 활용하여 S3 버킷의 존재 여부를 확인하는 방법을 알아봅니다. 예를 들어, Waiter를 사용해 S3에 Bucket_1이라는 버킷이 실제로 존재하는지 검사할 수 있습니다.

문제 해결 접근 방식 및 알고리즘

1단계 – boto3와 botocore 예외 클래스를 임포트하여 예외 상황을 처리할 수 있도록 준비합니다.

2단계 – 함수의 매개변수로 bucket_name을 사용합니다.

3단계 – boto3 라이브러리를 이용해 AWS 세션(Session)을 생성합니다.

4단계 – S3 서비스를 위한 AWS 클라이언트를 생성합니다.

5단계 – get_waiter 함수를 사용하여 bucket_exists에 대한 Waiter 객체를 생성합니다.

6단계 – Waiter 객체를 사용해 버킷의 존재 여부를 검증합니다. 기본 설정으로는 성공 상태에 도달할 때까지 5초 간격으로 계속 확인하며, 20번의 확인이 모두 실패하면 오류를 반환합니다. 다만, 사용자가 직접 폴링 간격(Delay)과 최대 시도 횟수(MaxAttempts)를 지정할 수도 있습니다.

7단계 – 이 함수의 반환값은 None입니다.

8단계 – 버킷 확인 과정에서 문제가 발생한 경우 일반 예외(generic exception)를 처리합니다.

예제 코드

다음 코드를 사용하면 Waiter를 통해 버킷의 존재 여부를 확인할 수 있습니다.

import boto3
from botocore.exceptions import ClientError

def use_waiters_check_bucket_exists(bucket_name):
    session = boto3.session.Session(profile_name='saml')
    s3_client = session.client('s3')
    try:
        waiter = s3_client.get_waiter('bucket_exists')
        waiter.wait(Bucket=bucket_name,
                    WaiterConfig={
                        'Delay': 2, 'MaxAttempts': 5})
        print('Bucket exists: ' + bucket_name)
    except ClientError as e:
        raise Exception("boto3 client error in use_waiters_check_bucket_exists: " + e.__str__())
    except Exception as e:
        raise Exception("Unexpected error in use_waiters_check_bucket_exists: " + e.__str__())

print(use_waiters_check_bucket_exists("Bucket_1"))
print(use_waiters_check_bucket_exists("Bucket_2"))

실행 결과

Bucket exists: Bucket_1
None

botocore.exceptions.WaiterError: Waiter BucketExists failed: Max
attempts exceeded
Exception: Unexpected error in use_waiters_check_bucket_exists: Waiter
BucketExists failed: Max attempts exceeded

결과 분석

Bucket_1의 경우, print 문장과 None이 출력됩니다. Waiter의 응답은 별도의 값을 반환하지 않기 때문에 None이 그대로 출력되는 것입니다.

반면 Bucket_2는 실제로 존재하지 않는 버킷이므로 예외가 발생합니다.

예외 메시지에서 "Max attempts exceeded", 즉 최대 시도 횟수를 초과했다는 내용을 확인할 수 있습니다. 이는 설정된 시도 횟수 내에 버킷이 존재한다는 성공 상태에 도달하지 못했음을 의미합니다.