S3对象版本回滚：优化效率与安全实践

import boto3
import logging
from operator import attrgetter

# 配置日志
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler())

def rollback_object_by_copy(bucket_name, object_key, target_version_id):
    """
    通过复制目标历史版本来实现S3对象回滚。
    此方法不会删除任何现有版本，而是将指定版本复制为当前最新版本。

    :param bucket_name: S3桶的名称。
    :param object_key: 要回滚的对象的键。
    :param target_version_id: 目标回滚版本的ID。
    :return: 回滚后新的活动版本ID。
    :raises KeyError: 如果目标版本ID未找到。
    :raises Exception: 其他S3操作错误。
    """
    s3_resource = boto3.resource('s3')
    bucket = s3_resource.Bucket(bucket_name)

    try:
        # 1. 验证目标版本是否存在（可选但推荐）
        # 尽管copy_from会在内部检查源版本，但提前检查可以提供更友好的错误信息
        # 注意：列出所有版本仍需使用Prefix，并在客户端过滤
        versions = bucket.object_versions.filter(Prefix=object_key)
        found_target_version = False
        for version in versions:
            if version.key == object_key and version.version_id == target_version_id:
                found_target_version = True
                break

        if not found_target_version:
            raise KeyError(f"错误: 版本ID '{target_version_id}' 未在对象 '{object_key}' 的版本列表中找到。")

        # 2. 构建CopySource参数
        copy_source = {
            'Bucket': bucket_name,
            'Key': object_key,
            'VersionId': target_version_id
        }

        # 3. 执行复制操作，使目标版本成为最新版本
        # copy_from方法将源对象（指定版本）复制到当前对象键下，创建新的版本
        bucket.Object(object_key).copy_from(CopySource=copy_source)

        # 4. 获取并打印新的当前版本ID
        # 注意：这里获取的是新创建的版本ID，它现在是活动版本
        current_active_version_id = bucket.Object(object_key).version_id
        logger.info(f"对象 '{object_key}' 已成功回滚到版本 '{target_version_id}'。")
        logger.info(f"当前活动版本ID为: {current_active_version_id}")

        return current_active_version_id

    except KeyError as e:
        logger.error(f"回滚失败: {e}")
        raise
    except Exception as e:
        logger.error(f"回滚对象 '{object_key}' 发生意外错误: {e}")
        raise

if __name__ == '__main__':
    # 请替换为您的实际S3桶名、对象键和目标版本ID
    my_bucket_name = 'scottedwards2000' 
    my_object_key = 'questions' 
    my_target_version_id = 'RQY0ebFXtUnm.A48N2I62CEmdu2QZGEO' 

    print(f"尝试将对象 '{my_object_key}' 回滚到版本 '{my_target_version_id}'...")
    try:
        new_active_version = rollback_object_by_copy(
            my_bucket_name, my_object_key, my_target_version_id
        )
        print(f"回滚操作成功完成。新的活动版本ID是: {new_active_version}")
    except Exception as e:
        print(f"回滚操作失败: {e}")