在DynamoDB中实现高效自增ID的两种策略

import boto3
from botocore.exceptions import ClientError
import logging

# 配置日志
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

def get_next_sequence_id(table_name, counter_pk_value="orderCounter", attribute_name="count"):
    """
    通过原子计数器获取下一个序列ID。

    Args:
        table_name (str): DynamoDB表名。
        counter_pk_value (str): 计数器项的Partition Key值。
        attribute_name (str): 存储计数器值的属性名。

    Returns:
        int: 下一个可用的序列ID。
    Raises:
        ClientError: DynamoDB操作失败。
        Exception: 其他意外错误。
    """
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table(table_name)

    try:
        # 原子递增计数器，并返回新值
        response = table.update_item(
            Key={'pk': counter_pk_value},
            UpdateExpression=f"ADD #{attribute_name} :val",
            ExpressionAttributeNames={f'#{attribute_name}': attribute_name},
            ExpressionAttributeValues={':val': 1},
            ReturnValues="UPDATED_NEW"
        )
        # 提取递增后的新值
        next_id = response['Attributes'][attribute_name]
        logger.info(f"Generated new sequence ID: {next_id}")
        return int(next_id)
    except ClientError as e:
        logger.error(f"Error updating atomic counter: {e}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error in get_next_sequence_id: {e}")
        raise

# 示例用法
if __name__ == "__main__":
    table_name = 'orders' # 假设你的表名为 'orders'
    try:
        next_order_id = get_next_sequence_id(table_name)

        # 使用新生成的ID创建新的订单项
        dynamodb = boto3.resource('dynamodb')
        orders_table = dynamodb.Table(table_name)
        orders_table.put_item(
            Item={
                'pk': str(next_order_id), # 将自增ID作为主键
                'deliveryMethod': 'expedited',
                'orderStatus': 'pending'
            }
        )
        logger.info(f"Order {next_order_id} created successfully.")
    except Exception as e:
        logger.error(f"Failed to process order: {e}")

import boto3
from boto3.dynamodb.conditions import Key
from botocore.exceptions import ClientError
import logging

# 配置日志
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

def get_next_issue_id(table_name, project_id):
    """
    通过排序键获取项目集合内的下一个序列ID。

    Args:
        table_name (str): DynamoDB表名。
        project_id (str): 项目的Partition Key值。

    Returns:
        int: 下一个可用的问题ID。
    Raises:
        ClientError: DynamoDB操作失败。
        Exception: 其他意外错误。
    """
    dynamodb = boto3.resource('dynamodb')
    client = dynamodb.Table(table_name)

    highest_issue_id = 0
    saved = False

    while not saved:
        try:
            # 查询特定项目（Partition Key）下最大的排序键（Sort Key）
            response = client.query(
                KeyConditionExpression=Key('pk').eq(project_id),
                ScanIndexForward=False, # 降序排列
                Limit=1 # 只取一个，即最大值
            )

            # 提取当前最大的序列号
            if response['Count'] > 0:
                highest_issue_id = int(response['Items'][0]['sk'])

            next_issue_id = highest_issue_id + 1
            logger.info(f"Attempting to write issue with ID: {next_issue_id} for project: {project_id}")

            # 使用条件写入，确保该序列号尚未存在
            # 这里使用 attribute_not_exists(sk) 更准确，因为 pk 必然存在
            client.put_item(
                Item={
                    'pk': project_id, 
                    'sk': next_issue_id, 
                    'priority': 'low',
                    'description': f'Issue {next_issue_id} for {project_id}'
                },
                ConditionExpression='attribute_not_exists(sk)' # 确保该排序键不存在
            )
            saved = True
            logger.info(f"Successfully created issue {next_issue_id} for project {project_id}.")
            return next_issue_id
        except client.meta.client.exceptions.ConditionalCheckFailedException:
            # 发生竞争条件，其他进程已写入相同ID，需要重试
            logger.warning(f"ConditionalCheckFailed for project {project_id}, retrying with next ID.")
            highest_issue_id = next_issue_id # 更新 highest_issue_id 为刚刚尝试的失败值，下次循环会在此基础上+1
            # 实际上，这里更安全的做法是重新查询 highest_issue_id，因为可能有多个并发写入
            # 但为了简化示例，我们直接在失败的ID上递增
            # 生产环境中，重新查询确保拿到最新的最高ID会更健壮
            # 如下所示：
            # highest_issue_id = 0 # 重置，重新查询
            # continue # 重新进入循环，再次查询最新最高ID
        except ClientError as e:
            logger.error(f"Error during DynamoDB operation: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error in get_next_issue_id: {e}")
            raise

# 示例用法
if __name__ == "__main__":
    table_name = 'projects' # 假设你的表名为 'projects'
    project_id_a = 'projectA'
    project_id_b = 'projectB'

    try:
        # 为 projectA 生成一个问题ID
        new_issue_id_a = get_next_issue_id(table_name, project_id_a)
        print(f"New issue ID for {project_id_a}: {new_issue_id_a}")

        # 为 projectB 生成一个问题ID
        new_issue_id_b = get_next_issue_id(table_name, project_id_b)
        print(f"New issue ID for {project_id_b}: {new_issue_id_b}")

        # 再次为 projectA 生成一个问题ID
        new_issue_id_a_2 = get_next_issue_id(table_name, project_id_a)
        print(f"Second new issue ID for {project_id_a}: {new_issue_id_a_2}")

    except Exception as e:
        logger.error(f"Failed to generate issue ID: {e}")