高效管理DynamoDB：基于排序键模式的批量数据删除策略

import boto3
from datetime import datetime
from typing import List, Dict

class DynamoDBBatchDeleter:
    """
    一个用于DynamoDB批量删除操作的类，专注于通过排序键模式进行高效删除。
    """
    def __init__(self, table_name: str, region_name: str = 'us-east-1'):
        """
        初始化DynamoDB客户端。

        Args:
            table_name (str): 目标DynamoDB表的名称。
            region_name (str): AWS区域名称。
        """
        self._dynamodb = boto3.resource('dynamodb', region_name=region_name)
        self._table = self._dynamodb.Table(table_name)
        print(f"初始化DynamoDBBatchDeleter，目标表: {table_name}")

    def batch_delete_old_data_by_sk_pattern(self, pk_value: str, date_threshold: datetime, sk_prefixes: List[int]) -> Dict[str, str]:
        """
        删除DynamoDB中符合指定分区键、排序键前缀和日期阈值条件的项目。

        Args:
            pk_value (str): 分区键的值 (例如 'abv')。
            date_threshold (datetime): 日期阈值。排序键中日期部分早于此日期的项目将被删除。
                                       例如，如果阈值为 '2023-12-12'，则 '2023-12-11' 及更早的日期将被删除。
            sk_prefixes (List[int]): 排序键中可能的整数前缀列表 (例如 [1, 2, ..., 30])。

        Returns:
            Dict[str, str]: 包含操作结果消息的字典。
        """
        try:
            items_to_delete = []
            # 将日期阈值格式化为 'YYYY-MM-DD' 字符串，用于排序键比较
            formatted_threshold_date = date_threshold.strftime('%Y-%m-%d')

            print(f"开始批量删除操作：pk='{pk_value}', 日期阈值 < '{formatted_threshold_date}'...")

            for prefix in sk_prefixes:
                # 构造用于排序键比较的上限字符串。
                # 例如，对于前缀 1，我们希望删除 sk < '1#2023-12-12' 的项目。
                sort_key_upper_bound = f"{prefix}#{formatted_threshold_date}"

                # 执行 Query 操作。KeyConditionExpression 结合了分区键和排序键条件。
                # 使用 Key().lt() 来查找小于指定排序键值的项目。
                query_response = self._table.query(
                    KeyConditionExpression=boto3.dynamodb.conditions.Key('pk').eq(pk_value) & \
                                         boto3.dynamodb.conditions.Key('sk').lt(sort_key_upper_bound)
                )

                # 处理查询结果的翻页（Pagination），确保获取所有匹配项目
                while True:
                    for item in query_response.get('Items', []):
                        items_to_delete.append({'pk': item['pk'], 'sk': item['sk']})

                    # 检查是否有更多结果页
                    if 'LastEvaluatedKey' in query_response:
                        query_response = self._table.query(
                            KeyConditionExpression=boto3.dynamodb.conditions.Key('pk').eq(pk_value) & \
                                                 boto3.dynamodb.conditions.Key('sk').lt(sort_key_upper_bound),
                            ExclusiveStartKey=query_response['LastEvaluatedKey']
                        )
                    else:
                        break # 没有更多页面，退出循环

            if not items_to_delete:
                print("未找到符合条件的老旧数据进行删除。")
                return {"message": "未找到符合条件的老旧数据进行删除。"}

            print(f"已找到 {len(items_to_delete)} 个项目待删除。正在启动批量写入器...")

            # 使用 batch_writer 进行高效批量删除。
            # boto3 的 batch_writer 会自动处理将删除请求分批 (每批最多25个项目)。
            with self._table.batch_writer() as batch:
                for item_key in items_to_delete:
                    batch.delete_item(Key=item_key)

            return {"message": f"老旧数据清理成功。共删除了 {len(items_to_delete)} 个项目。"}

        except Exception as e:
            print(f"批量删除过程中发生错误: {e}")
            raise # 重新抛出异常以便上层调用者处理

# 示例用法:
if __name__ == "__main__":
    # 请替换为您的实际表名和AWS区域
    TABLE_NAME = "YourDynamoDBTableName" 
    REGION = "your-aws-region" # 例如 'us-east-1', 'cn-north-1'

    deleter = DynamoDBBatchDeleter(TABLE_NAME, REGION)

    target_pk_value = 'abv'
    # 假设当前日期是 2023-12-12，我们想删除日期早于此日期的项目。
    # 所以阈值日期本身是 2023-12-12。所有日期为 '2023-12-11' 及更早的项目将被删除。
    threshold_date_for_deletion = datetime(2023, 12, 12) 

    # 根据问题描述，排序键前缀范围从 1 到 30
    sk_possible_prefixes = list(range(1, 31)) 

    try:
        result = deleter.batch_delete_old_data_by_sk_pattern(
            pk_value=target_pk_value,
            date_threshold=threshold_date_for_deletion,
            sk_prefixes=sk_possible_prefixes
        )
        print(result)
    except Exception as e:
        print(f"操作失败: {e}")