문제 상황
Python의 boto3 라이브러리를 사용해 AWS S3에 저장된 객체를 지정된 로컬 경로 또는 기본 경로로 다운로드하고, 이미 존재하는 파일을 덮어쓸지 여부(overwrite)까지 제어할 수 있는 함수를 만들어 보겠습니다. 예를 들어 S3의 Bucket_1/testfolder에 있는 test.zip 파일을 다운로드하는 상황을 가정합니다.
문제 해결 접근 방식 및 알고리즘
1단계 – boto3와 botocore의 예외 처리 모듈을 임포트합니다.
2단계 – pathlib에서 Path를 임포트해 파일명 존재 여부를 확인합니다.
3단계 – download_object_from_s3 함수는 s3_path, localpath, overwrite_existing_file이라는 세 개의 매개변수를 받습니다.
4단계 – s3_path가 s3://bucket_name/key 형식의 유효한 AWS 경로인지 검증합니다. localpath의 기본값은 None, overwrite_existing_file의 기본값은 True이며, 사용자가 원하는 로컬 경로를 직접 지정할 수도 있습니다.
5단계 – boto3 라이브러리로 AWS 세션을 생성합니다.
6단계 – S3를 위한 AWS 리소스를 생성합니다.
7단계 – S3 경로를 분리해 루트 버킷 이름과 다운로드할 객체 경로를 구분합니다.
8단계 – overwrite_existing_file이 False로 설정되어 있고 해당 파일이 지정된 로컬 경로에 이미 존재한다면 아무 작업도 수행하지 않습니다.
9단계 – 위 조건에 해당하지 않으면 객체를 다운로드합니다. localpath가 지정되어 있으면 해당 경로에, 없으면 기본 경로에 저장합니다.
10단계 – 응답 코드를 기반으로 예외를 처리해 파일이 정상적으로 다운로드되었는지 확인합니다.
11단계 – 다운로드 중 알 수 없는 오류가 발생한 경우 일반 예외로 처리합니다.
예제 코드
아래 코드를 사용하면 AWS S3에서 파일을 손쉽게 다운로드할 수 있습니다.
import boto3
from botocore.exceptions import ClientError
from pathlib import Path
def download_object_from_s3(s3path, localPath=None,
overwrite_existing_file=True):
if 's3://' not in s3path:
print('Given path is not a valid s3 path.')
raise Exception('Given path is not a valid s3 path.')
session = boto3.session.Session()
s3_resource = session.resource('s3')
s3_tokens = s3path.split('/')
bucket_name = s3_tokens[2]
object_path = ""
filename = s3_tokens[len(s3_tokens) - 1]
print('Filename: ' + filename)
if len(s3_tokens) > 4:
for tokn in range(3, len(s3_tokens) - 1):
object_path += s3_tokens[tokn] + "/"
object_path += filename
else:
object_path += filename
print('object: ' + object_path)
try:
if not overwrite_existing_file and Path.is_file(filename):
pass
else:
if localPath is None:
s3_resource.meta.client.download_file(bucket_name, object_path, filename)
else:
s3_resource.meta.client.download_file(bucket_name, object_path, localPath + '/' + filename)
print('Filename: ' + filename)
return filename
except ClientError as error:
if error.response['Error']['Code'] == '404':
print(s3path + " File not found: ")
raise Exception(s3path + " File not found: ")
except Exception as error:
print("Unexpected error in download_object function of s3 helper: " + error.__str__())
raise Exception("Unexpected error in download_object function of s3 helper: " + error.__str__())
#Download into default localpath
print(download_object_from_s3("s3://Bucket_1/testfolder/test.zip"))
#Download into given path
print(download_object_from_s3("s3://Bucket_1/testfolder/test.zip","C://AWS"))
#File doesn't exist in S3
print(download_object_from_s3("s3://Bucket_1/testfolder/abc.zip"))
실행 결과
#Download into default localpath Filename: test.zip object: testfolder/test.zip Filename: test.zip #Download into given path Filename: test.zip object: testfolder/test.zip Filename: test.zip #File doesn't exist in S3 Filename: abc.zip object: testfolder/abc.zip s3://Bucket_1/testfolder/abc.zip File not found: botocore.exceptions.ClientError: An error occurred (404) when calling the HeadObject operation: Not Found
참고 사항
기본 다운로드 경로는 이 함수가 작성된 디렉터리입니다. localpath를 별도로 지정하지 않으면 함수가 위치한 동일한 디렉터리에 파일이 저장됩니다.
예를 들어 이 함수가 S3_class라는 클래스 내부에 있고, 해당 클래스가 C://AWS/src/S3_class 경로에 있다면 test.zip 파일은 C://AWS/src/test.zip으로 다운로드됩니다.