DynamoDB基于排序键模式的批量删除策略

import boto3
from datetime import datetime, timedelta
from boto3.dynamodb.conditions import Key

class DynamoDBService:
    """
    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)

    def batch_delete_old_data_by_sk_pattern(self, 
                                            pk_value: str, 
                                            date_threshold_str: str,
                                            sk_prefix_range: tuple = (1, 30)) -> dict:
        """
        根据排序键模式执行批量删除操作。
        删除所有分区键为pk_value，且排序键的日期部分早于date_threshold_str的项。
        排序键格式假定为 "{integer}#{YYYY-MM-DD}"。

        Args:
            pk_value (str): 要操作的分区键值。
            date_threshold_str (str): 日期阈值字符串 (格式: 'YYYY-MM-DD')。
                                      所有日期早于此阈值的项将被删除。
            sk_prefix_range (tuple): 排序键前缀的整数范围 (min_int, max_int)。
                                     例如 (1, 30) 表示前缀从 '1#' 到 '30#'。

        Returns:
            dict: 包含操作结果的消息。

        Raises:
            Exception: 如果在操作过程中发生错误。
        """
        items_to_delete = []

        # 遍历所有可能的整数前缀
        for prefix_int in range(sk_prefix_range[0], sk_prefix_range[1] + 1):
            # 构建用于排序键比较的阈值字符串
            # 例如，如果 date_threshold_str 是 '2023-12-12'，
            # 那么对于前缀 '1#', 阈值是 '1#2023-12-12'。
            # 任何 '1#2023-12-11' 或更早的项都会被 Key('sk').lt() 匹配。
            threshold_sk = f"{prefix_int}#{date_threshold_str}"

            last_evaluated_key = None
            while True:
                query_params = {
                    'KeyConditionExpression': Key('pk').eq(pk_value) & Key('sk').lt(threshold_sk),
                    'ProjectionExpression': 'pk, sk'  # 仅获取主键属性以减少读取成本
                }
                if last_evaluated_key:
                    query_params['ExclusiveStartKey'] = last_evaluated_key

                try:
                    response = self._table.query(**query_params)
                    # 收集每个查询响应中的项
                    items_to_delete.extend([{'pk': item['pk'], 'sk': item['sk']} 
                                             for item in response.get('Items', [])])

                    last_evaluated_key = response.get('LastEvaluatedKey')
                    if not last_evaluated_key:
                        break # 没有更多项了，退出循环

                except Exception as e:
                    raise Exception(f"Error during Query for prefix {prefix_int}: {str(e)}")

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

        try:
            # 使用 batch_writer 批量删除收集到的项
            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:
            raise Exception(f"批量删除过程中发生错误: {str(e)}")

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

    # 初始化服务
    service = DynamoDBService(TABLE