如何使用boto3在DynamoDB中带条件更新字典列表（避免重复empId）

核心逻辑 #

要实现「不存在则创建列表，存在则仅当empId未重复时追加」的需求，关键是利用DynamoDB的ConditionExpression做前置检查，同时用if_not_exists处理列表未初始化的场景，还要确保所有表达式占位符都在ExpressionAttributeValues中定义（解决之前的ValidationException）。

完整代码示例 #

import boto3

# 初始化DynamoDB资源和表
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('你的表名称')

# 待添加的新对象（groupId是分区键，无需加入列表）
new_obj = {'groupId':101, 'empId': 1001, 'empName': 'Joe'}
# 提取要加入列表的内容（剔除分区键groupId）
new_list_item = {k: v for k, v in new_obj.items() if k != 'groupId'}

try:
    response = table.update_item(
        # 指定分区键
        Key={'groupId': new_obj['groupId']},
        # 更新逻辑：如果dataDetails不存在则初始化为空列表，再追加新元素
        UpdateExpression="SET dataDetails = if_not_exists(dataDetails, :empty_list) + :new_item",
        # 条件检查：列表中不存在empId匹配的元素才执行更新
        ConditionExpression="NOT EXISTS (dataDetails[*].empId[?@ = :emp_id])",
        # 定义所有表达式中用到的占位符值
        ExpressionAttributeValues={
            ':empty_list': [],
            ':new_item': [new_list_item],  # 必须包成列表，因为DynamoDB的+是列表拼接操作
            ':emp_id': new_obj['empId']
        },
        # 返回更新后的属性值（可选）
        ReturnValues='UPDATED_NEW'
    )
    print("元素追加成功", response)
except table.meta.client.exceptions.ConditionalCheckFailedException:
    # 条件不满足时触发：empId已存在于列表中
    print("该empId已存在，无需重复添加")
except Exception as e:
    print(f"更新失败: {str(e)}")

关键部分说明 #

1. UpdateExpression：

   • if_not_exists(dataDetails, :empty_list)：如果dataDetails属性不存在，就初始化为空列表；如果已存在则保留原列表。
   • + :new_item：将新元素（包成列表）拼接到原列表末尾。

2. ConditionExpression：

   • dataDetails[*].empId：投影出dataDetails列表中所有元素的empId字段。
   • [?@ = :emp_id]：过滤出empId等于目标值的元素。
   • NOT EXISTS(...)：仅当没有匹配的元素时，才允许执行更新操作。

3. 解决之前的ValidationException：
   之前的报错是因为UpdateExpression中用到了:empty_list但未在ExpressionAttributeValues中定义，现在明确声明:empty_list为[]即可解决。

内容的提问来源于stack exchange，提问作者sim