วิธีใช้ Boto3 Library ใน Python อัปโหลด Object เข้า S3 โดยใช้ AWS Resource

ข้อกำหนดปัญหา

ใช้ Boto3 Library ใน Python เพื่ออัปโหลด Object เข้า S3 เช่น อัปโหลด test.zip เข้า Bucket_1 ของ S3

ขั้นตอนการแก้ปัญหา

1. ขั้นตอนที่ 1 – นำเข้า boto3 และ botocore.exceptions เพื่อจัดการข้อยกเว้น
2. ขั้นตอนที่ 2 – จาก pathlib นำเข้า PurePosixPath เพื่อดึงชื่อไฟล์จาก Path
3. ขั้นตอนที่ 3 – s3_path และ filepath เป็นพารามิเตอร์ 2 ตัวในฟังก์ชัน upload_object_into_s3
4. ขั้นตอนที่ 4 – ตรวจสอบว่า s3_path ถูกส่งในรูปแบบ AWS คือ s3://bucket_name/key และ filepath เป็น Local Path เช่น C://users/filename
5. ขั้นตอนที่ 5 – สร้าง AWS Session ด้วย Boto3
6. ขั้นตอนที่ 6 – สร้าง AWS Resource สำหรับ S3
7. ขั้นตอนที่ 7 – แยก S3 Path และดำเนินการเพื่อแยก Root Bucket Name และ Key Path
8. ขั้นตอนที่ 8 – ดึงชื่อไฟล์จาก Filepath เต็มและเพิ่มเข้า S3 Key Path
9. ขั้นตอนที่ 9 – ใช้ฟังก์ชัน upload_fileobj เพื่ออัปโหลดไฟล์ Local เข้า S3
10. ขั้นตอนที่ 10 – ใช้ฟังก์ชัน wait_until_exists รอจนกว่าการดำเนินการจะเสร็จสิ้น
11. ขั้นตอนที่ 11 – จัดการ Exception ตาม Response Code เพื่อตรวจสอบว่าไฟล์ถูกอัปโหลดหรือไม่
12. ขั้นตอนที่ 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