วิธีใช้ Boto3 ดาวน์โหลด Object จาก S3 โดยใช้ AWS Resource

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

ใช้ไลบรารี Boto3 ใน Python เพื่อดาวน์โหลด Object จาก S3 ไปยัง Local Path ที่กำหนด/Path เริ่มต้น พร้อม Overwrite Existing File เป็น True เช่น ดาวน์โหลด test.zip จาก Bucket_1/testfolder ของ S3

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

ขั้นตอนที่ 1 – นำเข้า boto3 และ botocore.exceptions เพื่อจัดการข้อยกเว้น
ขั้นตอนที่ 2 – จาก pathlib นำเข้า Path เพื่อตรวจสอบชื่อไฟล์
ขั้นตอนที่ 3 – s3_path, localpath และ overwrite_existing_file เป็นพารามิเตอร์ 3 ตัวในฟังก์ชัน download_object_from_s3
ขั้นตอนที่ 4 – ตรวจสอบว่า s3_path ถูกส่งในรูปแบบ AWS คือ s3://bucket_name/key โดย Default localpath = None และ overwrite_existing_file = True ผู้ใช้สามารถส่งค่าเหล่านี้เพื่อดาวน์โหลดไปยัง Local Path ที่กำหนด
ขั้นตอนที่ 5 – สร้าง AWS Session ด้วย Boto3
ขั้นตอนที่ 6 – สร้าง AWS Resource สำหรับ S3
ขั้นตอนที่ 7 – แยก S3 Path และดำเนินการเพื่อแยก Root Bucket Name และ Object Path ที่จะดาวน์โหลด
ขั้นตอนที่ 8 – ตรวจสอบว่า overwrite_existing_file ตั้งเป็น False และไฟล์มีอยู่แล้วใน Local Path ที่กำหนด ในกรณีนี้ ไม่ทำอะไร
ขั้นตอนที่ 9 – มิฉะนั้น (หากเงื่อนไขใดเงื่อนไขหนึ่งไม่เป็นจริง) ให้ดาวน์โหลด Object หากมี localpath ให้ดาวน์โหลดที่นั่น มิฉะนั้นดาวน์โหลดไปยัง Default Path
ขั้นตอนที่ 10 – จัดการ Exception ตาม Response Code เพื่อตรวจสอบว่าไฟล์ถูกดาวน์โหลดหรือไม่
ขั้นตอนที่ 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

หมายเหตุ: Default Path สำหรับดาวน์โหลดคือ Directory ที่ฟังก์ชันนี้เขียนอยู่ ใน Directory เดียวกัน ไฟล์จะถูกดาวน์โหลดหากไม่ได้ระบุ Local Path

ตัวอย่าง หากฟังก์ชันนี้เขียนใน S3_class และ Class นี้อยู่ที่ C://AWS/src/S3_class แล้วไฟล์ test.zip จะถูกดาวน์โหลดไปที่ C://AWS/src/test.zip