如何使用boto3中断后重启S3大桶对象列表遍历

核心方案：利用S3分页标记实现精确断点续传 #

你提到的objects.all()无法直接跳过前N个对象，但可以通过S3 API原生支持的Continuation Token（对应v2版本的list_objects接口）实现断点续传，这是最可靠的方式——比跳过前N个更精准（S3对象按UTF-8字节序排序，跳过N个可能因分页逻辑出现偏差）。

具体实现步骤 #

1. 改用分页器（Paginator）控制遍历流程
   直接遍历objects.all()会自动处理分页，但无法捕获续传标记。改用boto3的分页器可以手动获取每一页的NextContinuationToken，用于断点记录。

2. 记录断点信息
   每次成功处理完一页对象后，将当前页返回的NextContinuationToken保存到本地文件或数据库中（比如JSON文件）。如果遍历中断，下次启动时读取这个token，就能从上次中断的位置继续。

3. 重启时从断点恢复
   重新通过STS获取新的临时凭证，初始化boto3客户端，然后将保存的ContinuationToken传入分页器，继续遍历。

代码示例 #

import boto3
import json
import os
from botocore.exceptions import ClientError

# 断点文件路径
CHECKPOINT_FILE = "s3_checkpoint.json"

def get_sts_credentials():
    # 替换为你的STS assume_role逻辑
    sts_client = boto3.client('sts')
    response = sts_client.assume_role(
        RoleArn='arn:aws:iam::123456789012:role/your-target-role',
        RoleSessionName='s3-bucket-list-session'
    )
    return response['Credentials']

def init_s3_client(credentials):
    return boto3.client(
        's3',
        aws_access_key_id=credentials['AccessKeyId'],
        aws_secret_access_key=credentials['SecretAccessKey'],
        aws_session_token=credentials['SessionToken']
    )

def load_continuation_token():
    try:
        with open(CHECKPOINT_FILE, 'r') as f:
            return json.load(f).get('continuation_token')
    except FileNotFoundError:
        return None

def save_continuation_token(token):
    with open(CHECKPOINT_FILE, 'w') as f:
        json.dump({'continuation_token': token}, f)

def traverse_large_bucket(bucket_name):
    credentials = get_sts_credentials()
    s3_client = init_s3_client(credentials)
    paginator = s3_client.get_paginator('list_objects_v2')
    
    # 加载上次的断点
    continuation_token = load_continuation_token()
    
    while True:
        try:
            # 构建分页请求参数
            page_params = {'Bucket': bucket_name}
            if continuation_token:
                page_params['ContinuationToken'] = continuation_token
            
            # 获取一页对象数据
            page = paginator.paginate(**page_params).build_full_result()
            objects = page.get('Contents', [])
            
            # 替换为你的对象信息收集逻辑
            for obj in objects:
                print(f"对象路径：{obj['Key']}，最后修改时间：{obj['LastModified']}")
            
            # 更新断点
            continuation_token = page.get('NextContinuationToken')
            if continuation_token:
                save_continuation_token(continuation_token)
            else:
                # 遍历完成，清理断点文件
                if os.path.exists(CHECKPOINT_FILE):
                    os.remove(CHECKPOINT_FILE)
                break
        
        except ClientError as e:
            error_code = e.response['Error']['Code']
            if error_code in ['ExpiredToken', 'InvalidToken']:
                # 凭证过期，重新获取凭证后继续
                print("临时凭证过期，重新获取...")
                credentials = get_sts_credentials()
                s3_client = init_s3_client(credentials)
                paginator = s3_client.get_paginator('list_objects_v2')
            else:
                # 其他异常直接抛出
                raise

if __name__ == "__main__":
    traverse_large_bucket('your-large-bucket-name')

额外优化建议 #

• 调整分页大小：在分页器中通过PageSize参数设置单次请求返回的对象数量（默认1000），避免单次请求数据量过大导致超时。
• 高频断点保存：建议每处理1-2页就保存一次断点，减少意外中断后的重复工作量。
• 扩展异常处理：除了凭证过期，还可以捕获网络波动、S3限流等异常，添加重试逻辑提升稳定性。

内容的提问来源于stack exchange，提问作者Daniel van der Maas