Đề bài – Sử dụng thư viện Boto3 trong Python để tải một đối tượng lên S3. Ví dụ: cách tải test.zip vào Bucket_1 của S3.
Phương pháp/Thuật toán giải quyết
Bước 1 – Import boto3 và các ngoại lệ từ botocore để xử lý lỗi.
Bước 2 – Từ pathlib, import PurePosixPath để lấy tên file từ đường dẫn.
Bước 3 – s3_path và filepath là hai tham số trong hàm upload_object_into_s3.
Bước 4 – Kiểm tra s3_path có đúng định dạng AWS s3://bucket_name/key và filepath là đường dẫn cục bộ C://users/filename không.
Bước 5 – Tạo phiên AWS (session) bằng thư viện boto3.
Bước 6 – Tạo resource AWS cho S3.
Bước 7 – Tách đường dẫn S3 và thực hiện các thao tác để tách tên bucket gốc và key path.
Bước 8 – Lấy tên file từ filepath hoàn chỉnh và thêm vào S3 key path.
Bước 9 – Sử dụng hàm upload_fileobj để tải file cục bộ lên S3.
Bước 10 – Sử dụng hàm wait_until_exists để chờ cho đến khi thao tác hoàn tất.
Bước 11 – Xử lý ngoại lệ dựa trên mã phản hồi để xác thực xem file đã được tải lên hay chưa.
Bước 12 – Xử lý ngoại lệ chung nếu có lỗi xảy ra khi tải lên file.
Ví dụ
Sử dụng mã sau để tải file lên 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 không phải là đường dẫn hợp lệ.' + filepath)
raise Exception('SourcePath không phải là đường dẫn hợp lệ.')
elif s3_path.find('s3://') == -1:
print('DestinationPath không phải là đường dẫn S3.' + s3_path)
raise Exception('DestinationPath không phải là đường dẫn hợp lệ.')
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:
# tải Entity từ đường dẫn cục bộ
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("Lỗi trong upload object function của s3 helper: " + error.__str__())
raise error
print(upload_object_into_s3('s3://Bucket_1/testfolder', 'c://test.zip'))
Kết quả
key_path:/testfolder/test.zip, target_bucket: Bucket_1 Object Uploaded Successfully