Computer >> 컴퓨터 >  >> 프로그래밍 >> Python

Python Boto3로 AWS S3에 객체(파일) 업로드하기 – 단계별 가이드와 코드 예제

문제 정의

Python의 Boto3 라이브러리를 사용하여 AWS S3에 객체(파일)를 업로드하는 방법을 알아보겠습니다. 예를 들어, 로컬 시스템에 있는 test.zip 파일을 S3의 Bucket_1 버킷에 업로드하려면 어떻게 해야 할까요?

사전 준비 사항

코드를 실행하기 전에 다음 두 가지가 준비되어 있어야 합니다.

  • boto3 설치: 터미널에서 pip install boto3 명령으로 라이브러리를 설치합니다.
  • AWS 자격 증명 설정: ~/.aws/credentials 파일 또는 환경 변수(AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)에 액세스 키가 등록되어 있어야 합니다.

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

1단계 – boto3와 botocore의 예외 클래스(ClientError)를 임포트하여 발생 가능한 오류를 처리할 수 있도록 준비합니다.

2단계 – pathlib 모듈에서 PurePosixPath를 임포트하여 전체 파일 경로에서 파일 이름만 손쉽게 추출합니다.

3단계 – 함수 upload_object_into_s3는 s3_path(S3 대상 경로)와 filepath(로컬 파일 경로)라는 두 개의 매개변수를 받습니다.

4단계 – s3_path가 s3://bucket_name/key 형식의 올바른 AWS 경로인지, filepath가 C://users/filename처럼 유효한 로컬 경로인지 검증합니다.

5단계 – boto3 라이브러리로 AWS 세션(Session)을 생성합니다.

6단계 – 생성한 세션을 통해 S3용 AWS 리소스 객체를 만듭니다.

7단계 – S3 경로를 분리(split)하여 루트 버킷 이름과 키(key) 경로를 구분합니다.

8단계 – 전체 파일 경로에서 파일 이름을 가져와 S3 키 경로에 추가합니다.

9단계 – upload_fileobj 함수를 사용하여 로컬 파일을 S3에 업로드합니다.

10단계 – wait_until_exists 함수로 업로드 작업이 완전히 끝날 때까지 대기합니다.

11단계 – 응답 코드를 기반으로 예외를 처리하여 파일이 정상적으로 업로드되었는지 확인합니다.

12단계 – 업로드 과정에서 문제가 발생한 경우 일반(generic) 예외를 처리합니다.

예제 코드

아래 코드를 사용하면 로컬 파일을 AWS S3에 업로드할 수 있습니다.

import boto3
from botocore.exceptions import ClientError
from pathlib import PurePosixPath

def upload_object_into_s3(s3_path, filepath):
    # 소스 경로가 S3 경로가 아닌지 확인
    if 's3://' in filepath:
        print('SourcePath is not a valid path.' + filepath)
        raise Exception('SourcePath is not a valid path.')
    elif s3_path.find('s3://') == -1:
        print('DestinationPath is not a s3 path.' + s3_path)
        raise Exception('DestinationPath is not a valid path.')

    # AWS 세션 및 S3 리소스 생성
    session = boto3.session.Session()
    s3_resource = session.resource('s3')

    # S3 경로를 분리하여 버킷 이름과 키 경로 추출
    tokens = s3_path.split('/')
    target_key = ""
    if len(tokens) > 3:
        for tokn in range(3, len(tokens)):
            if tokn == 3:
                target_key += tokens[tokn]
            else:
                target_key += "/" + tokens[tokn]
    target_bucket_name = tokens[2]

    # 로컬 파일 경로에서 파일 이름을 추출해 최종 키 경로 생성
    file_name = PurePosixPath(filepath).name
    if target_key != '':
        key_path = target_key + "/" + file_name
    else:
        key_path = file_name
    print("key_path: " + key_path, 'target_bucket: ' + target_bucket_name)

    try:
        # 로컬 경로의 파일을 S3로 업로드
        with open(filepath, "rb") as file:
            s3_resource.meta.client.upload_fileobj(file, target_bucket_name, key_path)
        try:
            # 업로드가 완료될 때까지 대기
            s3_resource.Object(target_bucket_name, key_path).wait_until_exists()
            file.close()
        except ClientError as error:
            error_code = int(error.response['Error']['Code'])
            if error_code == 412 or error_code == 304:
                print("Object didn't Upload Successfully ", target_bucket_name)
                raise error
        return "Object Uploaded Successfully"
    except Exception as error:
        print("Error in upload object function of s3 helper: " + error.__str__())
        raise error

print(upload_object_into_s3('s3://Bucket_1/testfolder', 'c://test.zip'))

실행 결과

key_path: testfolder/test.zip, target_bucket: Bucket_1
Object Uploaded Successfully

코드 핵심 포인트

upload_fileobj는 파일 경로가 아닌 파일 형태(file-like object)의 객체를 인자로 받기 때문에, open(filepath, "rb")로 연 바이너리 파일 객체를 그대로 전달합니다. 대용량 파일도 멀티파트 업로드로 자동 처리되므로 안정적입니다.

wait_until_exists는 S3 웨이터(Waiter) 기능으로, 객체가 실제로 버킷에 존재하게 될 때까지 폴링을 반복합니다. 이를 통해 업로드 요청이 성공했더라도 실제 저장이 완료되지 않은 상태에서 다음 작업으로 넘어가는 실수를 방지할 수 있습니다.

예외 처리 부분에서는 ClientError의 응답 코드를 확인합니다. 특히 412(사전 조건 실패)나 304(수정되지 않음) 코드가 반환되면 업로드가 제대로 되지 않은 것으로 판단하고 오류를 다시 발생시켜 호출자에게 알립니다. 그 외 모든 예외는 최상위 except 블록에서 일괄 처리됩니다.