문제 설명 − Python에서 Boto3 라이브러리를 사용하여 객체를 S3에 업로드합니다. 예를 들어 test.zip을 S3의 Bucket_1에 업로드하는 방법입니다.
이 문제를 해결하기 위한 접근 방식/알고리즘
1단계 − boto3 및 botocore 예외를 가져와 예외를 처리합니다.
2단계 − pathlib에서 , PurePosixPath를 가져와 경로에서 파일 이름 검색
3단계 - s3_path 및 파일 경로 upload_object_into_s3 함수의 두 매개변수입니다.
4단계 − s3_path 확인 s3://bucket_name/key로 AWS 형식으로 전달됩니다. 및 파일 경로 로컬 경로로 C://users/filename
5단계 − boto3 라이브러리를 사용하여 AWS 세션을 생성합니다.
6단계 − S3용 AWS 리소스를 생성합니다.
7단계 − S3 경로를 분할하고 루트 버킷 이름과 키 경로를 분리하는 작업을 수행합니다.
8단계 − 전체 파일 경로에 대한 파일 이름을 가져와 S3 키 경로에 추가합니다.
9단계 − 이제 upload_fileobj 기능을 사용하세요. 로컬 파일을 S3에 업로드합니다.
10단계 − wait_until_exists 기능 사용 작업이 완료될 때까지 기다리십시오.
11단계 − 응답 코드를 기반으로 예외를 처리하여 파일 업로드 여부를 확인합니다.
12단계 − 파일을 업로드하는 동안 문제가 발생한 경우 일반 예외 처리
예시
다음 코드를 사용하여 AWS S3에 파일을 업로드하십시오 -
import boto3
from botocore.exceptions import ClientError
from pathlib import PurePosixPath
def upload_object_into_s3(s3_path, filepath):
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.')
session = boto3.session.Session()
s3_resource = session.resource('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 != '':
target_key.strip()
key_path = target_key + "/" + file_name
else:
key_path = file_name
print(("key_path: " + key_path, 'target_bucket: ' + target_bucket_name))
try:
# uploading Entity from local path
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