home

หน้าแรก
 Computer >> บทช่วยสอนคอมพิวเตอร์ >  >> การเขียนโปรแกรม >> Python

Python Boto3: ดาวน์โหลดไฟล์จาก AWS S3 คู่มือครบวงจร

Boto3 เป็น AWS SDK สำหรับ Python ใช้งาน S3, EC2, DynamoDB และบริการ AWS อื่นๆ บทความนี้เน้นการ Download ไฟล์จาก S3 Bucket

Prerequisites

pip install boto3

AWS Credentials Setup

ตั้งค่า Credentials หนึ่งในวิธี:

  • aws configure (CLI)
  • Environment Variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
  • IAM Role (ถ้ารันบน EC2/Lambda)
  • ~/.aws/credentials file

1. ดาวน์โหลดไฟล์เดี่ยว: download_file()

import boto3

BUCKET_NAME = 'my-bucket'
OBJECT_KEY = 'path/to/file.json'
LOCAL_FILENAME = 'downloaded.json'

def download_single_file():
    s3 = boto3.client('s3')
    s3.download_file(BUCKET_NAME, OBJECT_KEY, LOCAL_FILENAME)
    print(f"Downloaded to {LOCAL_FILENAME}")

download_single_file()

Alternative: download_fileobj() (Stream to Memory)

import boto3
import io

def download_to_memory():
    s3 = boto3.client('s3')
    buffer = io.BytesIO()
    s3.download_fileobj(BUCKET_NAME, OBJECT_KEY, buffer)
    buffer.seek(0)
    content = buffer.read()
    return content

2. ดาวน์โหลดไฟล์ทั้งหมดใน Bucket

import boto3

def download_all_files(bucket_name: str, local_dir: str = '.'):
    s3 = boto3.resource('s3')
    bucket = s3.Bucket(bucket_name)
    
    for obj in bucket.objects.all():
        # สร้างโฟลเดอร์ถ้ายังไม่มี
        local_path = f"{local_dir}/{obj.key}"
        import os
        os.makedirs(os.path.dirname(local_path), exist_ok=True)
        
        bucket.download_file(obj.key, local_path)
        print(f"Downloaded: {obj.key}")

# ใช้งาน
download_all_files('my-bucket', './downloads')

3. ดาวน์โหลดไฟล์ใน Sub-folder (Prefix Filter)

import boto3
import os

def download_folder(bucket_name: str, prefix: str, local_dir: str = '.'):
    """
    prefix: เช่น 'pets/cats/' (ต้องลงท้ายด้วย /)
    """
    s3 = boto3.resource('s3')
    bucket = s3.Bucket(bucket_name)
    
    # Filter ด้วย Prefix
    objects = bucket.objects.filter(Prefix=prefix)
    
    for obj in objects:
        if obj.key.endswith('/'):  # ข้าม "folder" marker
            continue
            
        # Relative path จาก prefix
        relative_path = obj.key[len(prefix):]
        local_path = os.path.join(local_dir, relative_path)
        
        os.makedirs(os.path.dirname(local_path), exist_ok=True)
        bucket.download_file(obj.key, local_path)
        print(f"Downloaded: {obj.key} -> {local_path}")

# ใช้งาน: ดาวน์โหลดจาก bucket 'images' โฟลเดอร์ 'pets/cats/'
download_folder('images', 'pets/cats/', './cat_images')

4. Pagination สำหรับ Bucket ใหญ่ (List Objects V2)

def download_all_paginated(bucket_name: str, prefix: str = '', max_keys: int = 1000):
    s3 = boto3.client('s3')
    paginator = s3.get_paginator('list_objects_v2')
    
    for page in paginator.paginate(Bucket=bucket_name, Prefix=prefix):
        if 'Contents' not in page:
            continue
            
        for obj in page['Contents']:
            key = obj['Key']
            if key.endswith('/'):
                continue
            # download logic here
            s3.download_file(bucket_name, key, key.split('/')[-1])

Error Handling

import botocore.exceptions

def safe_download(bucket, key, local_path):
    s3 = boto3.client('s3')
    try:
        s3.download_file(bucket, key, local_path)
    except botocore.exceptions.ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code == '404':
            print(f"File not found: {key}")
        elif error_code == '403':
            print(f"Access denied: {key}")
        else:
            print(f"Error: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")

References