高效管理S3对象版本：非破坏性回滚策略与实践

import boto3
import logging
from operator import attrgetter

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

def rollback_object_destructive(bucket_name, object_key, target_version_id):
    """
    通过删除指定版本之后的所有版本来回滚S3对象。
    这种方法是破坏性的，会永久删除较新的版本。

    :param bucket_name: S3桶名称。
    :param object_key: 要回滚的对象键。
    :param target_version_id: 要回滚到的目标版本ID。
    """
    s3_resource = boto3.resource('s3')
    bucket = s3_resource.Bucket(bucket_name)

    # 获取所有以object_key为前缀的版本，并按修改时间降序排列
    # 注意：filter(Prefix=object_key) 可能包含其他以该前缀命名的对象
    versions = sorted(
        bucket.object_versions.filter(Prefix=object_key),
        key=attrgetter("last_modified"),
        reverse=True,
    )

    # 在客户端进行精确过滤，确保只处理目标对象的版本
    filtered_versions = [v for v in versions if v.key == object_key]

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

    if target_version_id in [ver.version_id for ver in filtered_versions]:
        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"对象 '{object_key}' 的当前活动版本是: {current_active_version_id}")
    else:
        raise KeyError(
            f"版本ID '{target_version_id}' 未在对象 '{object_key}' 的版本列表中找到。"
        )

# 示例调用 (请替换为您的桶名、对象键和版本ID)
# if __name__ == '__main__':
#    mybucket_name = 'your-s3-bucket-name'
#    my_object_key = 'your-object-key'
#    my_target_version_id = 'your-target-version-id'
#    rollback_object_destructive(mybucket_name, my_object_key, my_target_version_id)

import boto3
import logging

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

def rollback_object_non_destructive(bucket_name, object_key, target_version_id):
    """
    通过复制目标版本来非破坏性地回滚S3对象。
    此方法不会删除任何现有版本，而是创建一个新的版本作为当前版本。

    :param bucket_name: S3桶名称。
    :param object_key: 要回滚的对象键。
    :param target_version_id: 要回滚到的目标版本ID。
    """
    s3_resource = boto3.resource('s3')
    bucket = s3_resource.Bucket(bucket_name)

    # 构造CopySource，指定源桶、源对象键和要复制的特定版本ID
    copy_source = {
        'Bucket': bucket_name,
        'Key': object_key,
        'VersionId': target_version_id
    }

    try:
        # 执行复制操作，将目标版本复制为当前对象的新版本
        # S3会自动创建一个新的版本，其内容与target_version_id一致
        bucket.Object(object_key).copy_from(CopySource=copy_source)

        # 验证当前活动版本
        current_active_version_id = bucket.Object(object_key).version_id
        print(f"对象 '{object_key}' 已非破坏性回滚到版本 '{target_version_id}'。")
        print(f"当前活动版本是新的版本ID: {current_active_version_id}")
        logger.info(f"成功将对象 '{object_key}' 回滚到版本 '{target_version_id}'. 新的活动版本ID: {current_active_version_id}")

    except s3_resource.meta.client.exceptions.ClientError as e:
        if e.response['Error']['Code'] == 'NoSuchKey':
            logger.error(f"指定版本ID '{target_version_id}' 或对象键 '{object_key}' 不存在。")
            raise KeyError(f"版本ID '{target_version_id}' 或对象键 '{object_key}' 不存在。")
        else:
            logger.error(f"回滚对象 '{object_key}' 时发生错误: {e}")
            raise

# 示例调用 (请替换为您的桶名、对象键和版本ID)
if __name__ == '__main__':
   # 假设您有一个名为 'my-test-bucket-123' 的桶
   # 并且其中有一个对象 'my-document.txt' 存在多个版本
   my_bucket_name = 'scottedwards2000' 
   my_object_key = 'questions'

   # 假设这是您希望回滚到的一个历史版本ID
   # 您需要提前知道这个版本ID，可以通过 list_object_versions 获取
   my_target_version_id = 'RQY0ebFXtUnm.A48N2I62CEmdu2QZGEO' 

   print(f"尝试将对象 '{my_object_key}' 回滚到版本 '{my_target_version_id}'...")
   rollback_object_non_destructive(my_bucket_name, my_object_key, my_target_version_id)

   # 再次确认当前版本
   s3_resource = boto3.resource('s3')
   current_version_id = s3_resource.Bucket(my_bucket_name).Object(my_object_key).version_id
   print(f"回滚后，'{my_object_key}' 的最新版本ID是: {current_version_id}")