DynamoDB 自增ID实现指南

import boto3

# 初始化 DynamoDB 资源
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('orders') # 假设你的表名为 'orders'

def get_next_order_id():
    try:
        # 对名为 'orderCounter' 的项进行原子增量操作
        # 'pk' 是分区键，这里假设为 'orderCounter'
        # 'count' 是存储计数值的属性
        response = table.update_item(
            Key={'pk': 'orderCounter'},
            UpdateExpression="ADD #cnt :val",
            ExpressionAttributeNames={'#cnt': 'count'},
            ExpressionAttributeValues={':val': 1},
            ReturnValues="UPDATED_NEW" # 返回更新后的新值
        )
        # 提取更新后的计数值
        next_order_id = response['Attributes']['count']
        return next_order_id
    except Exception as e:
        print(f"获取下一个订单ID时发生错误: {e}")
        raise

# 使用新生成的ID创建新订单项
try:
    next_order_id = get_next_order_id()
    print(f"生成的下一个订单ID: {next_order_id}")

    # 使用这个新ID插入新项
    table.put_item(
        Item={
            'pk': str(next_order_id), # 将ID转换为字符串作为分区键
            'deliveryMethod': 'expedited',
            'orderDate': '2023-10-27'
        }
    )
    print(f"订单 {next_order_id} 已成功创建。")

except Exception as e:
    print(f"创建订单时发生错误: {e}")

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

# 初始化 DynamoDB 资源
dynamodb = boto3.resource('dynamodb')
client = dynamodb.Table('projects') # 假设你的表名为 'projects'

PROJECT_ID = 'projectA' # 示例项目ID

def create_new_issue(project_id, priority):
    highest_issue_id = 0
    saved = False

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

            # 如果存在项，则获取最大的 issue ID
            if response['Count'] > 0:
                highest_issue_id = int(response['Items'][0]['sk'])

            # 尝试使用下一个序列值写入新项
            new_issue_id = highest_issue_id + 1
            client.put_item(
                Item={
                    'pk': project_id,
                    'sk': new_issue_id, # 排序键作为 issue ID
                    'priority': priority
                },
                # 条件表达式：只有当该主键组合（pk+sk）不存在时才写入成功
                ConditionExpression='attribute_not_exists(pk) AND attribute_not_exists(sk)'
            )
            saved = True
            print(f"项目 {project_id} 的新问题 {new_issue_id} 已成功创建。")
            return new_issue_id
        except ClientError as e:
            # 如果是条件检查失败，说明发生了竞态条件，需要重试
            if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
                print(f"竞态条件发生，项目 {project_id} 的问题ID {highest_issue_id + 1} 已被占用，重试...")
                # 重新查询或直接递增 highest_issue_id 并重试
                # 这里简单地递增，实际生产中更推荐重新查询以获取最新的最大值
                highest_issue_id = highest_issue_id + 1 # 简单递增，然后循环重试
            else:
                print(f"创建问题时发生其他错误: {e}")
                raise
        except Exception as e:
            print(f"创建问题时发生意外错误: {e}")
            raise

# 调用函数创建新问题
try:
    new_id = create_new_issue(PROJECT_ID, 'low')
    print(f"最终创建的问题ID: {new_id}")
except Exception as e:
    print(f"主程序错误: {e}")