调用DynamoDB Query报ValidationException 如何在Python Lambda中查询表最大ID

错误原因 #

DynamoDB的Query操作是针对单个分区键的定向查询，强制要求传入非空的KeyConditionExpression指定分区键匹配规则，你将该参数留空触发了参数校验错误。你参考的最大ID查询方案仅适用于「相同分区键下、按排序键排序取最大值」的场景，无法直接用来查询全表最大ID。

可行实现方案 #

方案1：小表用Scan查询全表最大ID（仅适用于数据量低于1000条的测试场景，大表Scan会消耗大量读容量且延迟高） #

修正后的完整代码如下：

import json
import boto3

TABLE_NAME = 'user-profiles'

dynamo_DB = boto3.resource('dynamodb')    

def lambda_handler(event, context):
    user_id = event['user_id']
    email = event['email']
    bvn = event['bvn']
    password = event['password']
    phone = event['phone']
    gender = event['gender']

    output = ''

    if len(user_id) > 1 and len(password) > 5:
        try:
            table = dynamo_DB.Table(TABLE_NAME)
            # 替换原来的Query为Scan，添加ProjectionExpression只查id字段减少开销
            scan_resp = table.scan(
                ProjectionExpression='id'
            )
            # 遍历所有id取最大值，空表默认从1开始
            max_id = max(int(item['id']) for item in scan_resp['Items']) if scan_resp['Items'] else 0
            new_id = max_id + 1

            Item = {
                'id': str(new_id),
                'profile-id': str(new_id),
                'user_id': user_id,
                'email': email,
                'bvn': bvn,
                'password': password,
                'phone': phone,
                'gender': gender
            }
            table.put_item(Item=Item)

            output += 'Data Inserted To Dynamodb Successfully'
        except Exception as e:
            output += 'error with dynamo registration ' + str(e)

    else:
        output += 'invalid user or password entered, this is ' \
                  'what i received:\nusername: ' \
                  + str(user_id) + '\npassword: ' + str(password)

    return {
        "statusCode": 200,
        "body": json.dumps({
            "message": output,
        }),
    }

方案2：同分区键下用Query取最大值（生产推荐，需要表结构满足：分区键为固定值字段、id为排序键） #

如果你的表所有用户记录的分区键都是同一个固定值（比如分区键名称为record_type，值固定为user_profile），且id是排序键，就可以用高性能的Query实现，仅需替换原来的查询逻辑：

# 替换原有query代码段
values = table.query(
    KeyConditionExpression=boto3.dynamodb.conditions.Key('record_type').eq('user_profile'),
    ScanIndexForward=False,
    Limit=1
)
max_id = int(values['Items'][0]['id']) if values['Items'] else 0
new_id = max_id + 1

额外注意事项 #

• 当前你用的ID自增逻辑在高并发场景下会出现ID冲突，多个请求同时查到相同的max_id，写入时会覆盖旧数据。生产环境建议用DynamoDB原子计数器、或者直接用UUID作为主键，不要用自增ID。
• 不要在数据量超过1000条的生产表用Scan操作，会产生极高的读容量消耗和延迟。

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