S3对象版本回滚：精确键过滤与高效策略

import boto3
import logging
from operator import attrgetter

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

def rollback_object(bucket, object_key, version_id):
    """
    通过删除指定回滚版本之后的所有版本，将对象回滚到早期版本。

    :param bucket: 包含要回滚对象的桶。
    :param object_key: 要回滚的对象键。
    :param version_id: 要回滚到的版本ID。
    """
    # 获取所有以 object_key 为前缀的版本。注意：这里只能使用 Prefix。
    # 版本必须按 last_modified 日期排序，因为删除标记通常在列表末尾。
    versions = sorted(
        bucket.object_versions.filter(Prefix=object_key),
        key=attrgetter("last_modified"),
        reverse=True, # 降序排序，最新的版本在前
    )

    # 客户端精确过滤：确保只处理与 object_key 完全匹配的版本。
    # 这一步是必要的，因为 Prefix 过滤可能包含其他以相同前缀开头的对象。
    filtered_versions = [v for v in versions if v.key == object_key]

    logger.debug(
        "获取到的版本列表:\n%s",
        "\n".join(
            [
                f"\t{version.version_id}, 最后修改时间 {version.last_modified}"
                for version in filtered_versions
            ]
        ),
    )

    if version_id in [ver.version_id for ver in filtered_versions]:
        print(f"正在回滚到版本 {version_id}")
        for version in filtered_versions:
            # 遍历版本，如果版本ID不是目标版本，则删除。
            # 由于是降序排序，第一个遇到的非目标版本就是最新的，依次删除直到目标版本。
            if version.version_id != version_id:
                version.delete()
                print(f"已删除版本 {version.version_id}")
            else:
                # 找到目标版本，停止删除
                break

        # 打印当前活动版本
        print(f"当前活动版本为 {bucket.Object(object_key).version_id}")
    else:
        raise KeyError(
            f"{version_id} 未在对象 {object_key} 的版本列表中找到。"
        )

if __name__ == '__main__':
   # 示例用法，请替换为您的桶名、对象键和版本ID
   # mybucket = boto3.resource('s3').Bucket('您的桶名')
   # result = rollback_object(mybucket, '您的对象键', '您的版本ID')
   # print(result)
   pass # 避免直接运行示例代码，需要用户手动配置

import boto3

def rollback_object_by_copy(bucket_name, object_key, target_version_id):
    """
    通过将指定历史版本复制到当前对象键，实现S3对象版本回滚。
    这会创建一个新的版本，其内容与目标历史版本相同。

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

    try:
        # 构造源对象的CopySource
        copy_source = {
            'Bucket': bucket_name,
            'Key': object_key,
            'VersionId': target_version_id
        }

        # 将目标版本复制到当前对象键。这会创建一个新的版本。
        bucket.Object(object_key).copy_from(CopySource=copy_source)

        print(f"对象 '{object_key}' 已成功回滚到版本 '{target_version_id}'。")
        # 获取并打印新的当前版本ID
        current_version_id = bucket.Object(object_key).version_id
        print(f"当前活动版本ID为: {current_version_id}")

    except Exception as e:
        print(f"回滚失败: {e}")
        raise

if __name__ == '__main__':
    # 示例用法，请替换为您的桶名、对象键和版本ID
    # bucket_name = 'scottedwards2000'
    # object_key = 'questions'
    # target_version = 'RQY0ebFXtUnm.A48N2I62CEmdu2QZGEO'
    # rollback_object_by_copy(bucket_name, object_key, target_version)
    pass # 避免直接运行示例代码