如何使用Boto3从S3下载整个文件夹？已掌握单个文件下载方法

嘿，好问题！首先得明确一点：S3其实并没有真正的「文件夹」概念——所谓的文件夹只是带有特定前缀的对象集合而已。不过别担心，我们可以通过遍历前缀下的所有对象，逐个下载来实现「下载整个文件夹」的效果。

方法一：基础遍历下载（适合小批量文件） #

这是最直接的方式，通过列出指定前缀下的所有对象，循环调用download_file下载，同时自动保留本地的目录结构。

示例代码：

import boto3
import os

def download_s3_folder(bucket_name, s3_prefix, local_dir):
    # 初始化S3客户端
    s3 = boto3.client('s3')
    
    # 分页列出前缀下的所有对象（避免单次返回数量过多）
    paginator = s3.get_paginator('list_objects_v2')
    page_iterator = paginator.paginate(Bucket=bucket_name, Prefix=s3_prefix)
    
    for page in page_iterator:
        if 'Contents' not in page:
            continue  # 如果前缀下没有对象，直接跳过
        
        for obj in page['Contents']:
            s3_key = obj['Key']
            # 跳过前缀本身（如果是S3里的空“文件夹”对象）
            if s3_key == s3_prefix:
                continue
            
            # 构建本地文件路径：去掉前缀，保留剩余的目录层级
            relative_path = os.path.relpath(s3_key, s3_prefix)
            local_file_path = os.path.join(local_dir, relative_path)
            
            # 创建本地目录（如果不存在）
            os.makedirs(os.path.dirname(local_file_path), exist_ok=True)
            
            # 下载文件
            s3.download_file(bucket_name, s3_key, local_file_path)
            print(f"已完成下载：{s3_key} -> {local_file_path}")

# 调用示例
if __name__ == "__main__":
    BUCKET_NAME = "你的存储桶名称"
    S3_FOLDER_PREFIX = "path/to/your/folder/"  # 注意末尾的斜杠，确保只匹配该前缀下的对象
    LOCAL_DOWNLOAD_DIR = "./本地下载目录"
    
    download_s3_folder(BUCKET_NAME, S3_FOLDER_PREFIX, LOCAL_DOWNLOAD_DIR)

方法二：并发下载（适合大批量文件） #

如果要下载的文件数量很多，用S3Transfer可以开启并发下载，提升效率：

import boto3
from boto3.s3.transfer import TransferConfig, S3Transfer
import os

def download_s3_folder_concurrent(bucket_name, s3_prefix, local_dir, max_concurrent=10):
    s3 = boto3.client('s3')
    
    # 配置并发传输参数
    config = TransferConfig(max_concurrency=max_concurrent)
    transfer = S3Transfer(client=s3, config=config)
    
    paginator = s3.get_paginator('list_objects_v2')
    page_iterator = paginator.paginate(Bucket=bucket_name, Prefix=s3_prefix)
    
    for page in page_iterator:
        if 'Contents' not in page:
            continue
        
        for obj in page['Contents']:
            s3_key = obj['Key']
            if s3_key == s3_prefix:
                continue
            
            relative_path = os.path.relpath(s3_key, s3_prefix)
            local_file_path = os.path.join(local_dir, relative_path)
            os.makedirs(os.path.dirname(local_file_path), exist_ok=True)
            
            transfer.download_file(bucket_name, s3_key, local_file_path)
            print(f"已完成下载：{s3_key} -> {local_file_path}")

# 调用示例
if __name__ == "__main__":
    BUCKET_NAME = "你的存储桶名称"
    S3_FOLDER_PREFIX = "path/to/your/folder/"
    LOCAL_DOWNLOAD_DIR = "./本地下载目录"
    
    download_s3_folder_concurrent(BUCKET_NAME, S3_FOLDER_PREFIX, LOCAL_DOWNLOAD_DIR)

关键注意事项 #

• 权限：确保你的AWS凭证拥有s3:ListBucket（列出存储桶内容）和s3:GetObject（下载对象）的权限。
• 前缀斜杠：前缀末尾的/必须加上，比如my-folder/只会匹配该“文件夹”下的对象，而my-folder会匹配所有以my-folder开头的对象（包括my-folderfile.txt这种非目标文件）。
• 嵌套目录：代码会自动创建对应的本地嵌套目录，无需手动处理路径问题。

内容的提问来源于stack exchange，提问作者El Fadel Anas