优化Amazon S3对象版本回滚策略：从前缀过滤到高效复制

import boto3
import logging
from operator import attrgetter

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler())

def rollback_object_by_deletion(bucket_name, object_key, target_version_id):
    """
    通过删除较新版本来回滚S3对象。
    此方法会删除目标版本之后的所有版本。
    """
    s3_resource = boto3.resource('s3')
    bucket = s3_resource.Bucket(bucket_name)

    # 1. 使用Prefix获取版本列表（可能包含非目标对象的版本）
    # 必须按last_modified日期排序，因为删除标记可能在列表末尾
    all_versions = sorted(
        bucket.object_versions.filter(Prefix=object_key),
        key=attrgetter("last_modified"),
        reverse=True, # 降序排列，最新版本在前
    )

    # 2. 客户端精确过滤，确保只处理目标对象
    filtered_versions = [v for v in all_versions if v.key == object_key]

    if not filtered_versions:
        raise KeyError(f"未找到对象 {object_key} 的任何版本。")

    logger.debug(
        "获取到对象 %s 的版本:\n%s",
        object_key,
        "\n".join(
            [
                f"\t版本ID: {version.version_id}, 最后修改时间: {version.last_modified}, 是否删除标记: {version.is_latest}"
                for version in filtered_versions
            ]
        ),
    )

    # 3. 检查目标版本是否存在，并执行删除操作
    if target_version_id not in [ver.version_id for ver in filtered_versions]:
        raise KeyError(
            f"版本ID {target_version_id} 未在对象 {object_key} 的版本列表中找到。"
        )

    print(f"开始回滚对象 {object_key} 到版本 {target_version_id}")
    for version in filtered_versions:
        if version.version_id != target_version_id:
            # 迭代删除每个比目标版本新的版本
            version.delete()
            print(f"已删除版本 {version.version_id}")
        else:
            # 达到目标版本，停止删除
            break

    # 验证当前活动版本
    current_active_version_id = bucket.Object(object_key).version_id
    print(f"回滚完成。当前活动版本为 {current_active_version_id}")
    return current_active_version_id

# 示例用法（请替换为您的桶名、对象键和版本ID）
if __name__ == '__main__':
   # mybucket_name = 'your-s3-bucket-name'
   # my_object_key = 'your-object-key'
   # my_target_version_id = 'your-target-version-id'
   # try:
   #     rollback_object_by_deletion(mybucket_name, my_object_key, my_target_version_id)
   # except KeyError as e:
   #     print(f"错误: {e}")
   pass

import boto3
import logging

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler())

def rollback_object_by_copy(bucket_name, object_key, target_version_id):
    """
    通过复制目标版本来回滚S3对象。
    此方法会将指定的旧版本复制为当前最新版本，不删除任何历史版本。
    """
    s3_resource = boto3.resource('s3')
    bucket = s3_resource.Bucket(bucket_name)

    # 构造源对象信息，包括桶名、对象键和目标版本ID
    copy_source = {
        'Bucket': bucket_name,
        'Key': object_key,
        'VersionId': target_version_id
    }

    try:
        # 执行复制操作，目标是同一个对象键，这将创建一个新版本
        # 新创建的版本内容与target_version_id相同，并成为最新的活动版本
        bucket.copy(copy_source, object_key)

        # 验证当前活动版本
        current_active_version_id = bucket.Object(object_key).version_id
        print(f"对象 {object_key} 已成功回滚到版本 {target_version_id}。")
        print(f"当前活动版本为 {current_active_version_id}")
        return current_active_version_id
    except s3_resource.meta.client.exceptions.ClientError as e:
        if e.response['Error']['Code'] == 'NoSuchVersion':
            raise KeyError(f"版本ID {target_version_id} 未在对象 {object_key} 的版本列表中找到。")
        else:
            raise

# 示例用法（请替换为您的桶名、对象键和版本ID）
if __name__ == '__main__':
   mybucket_name = 'scottedwards2000' # 替换为您的S3桶名
   my_object_key = 'questions' # 替换为您的对象键
   my_target_version_id = 'RQY0ebFXtUnm.A48N2I62CEmdu2QZGEO' # 替换为您要回滚到的目标版本ID
   try:
       rollback_object_by_copy(mybucket_name, my_object_key, my_target_version_id)
   except KeyError as e:
       print(f"错误: {e}")
   except Exception as e:
       print(f"发生未知错误: {e}")