使用Lambda结合随机数生成在DynamoDB创建条目时遇payload解析错误

一、修复json.loads(event['body'])报错 #

报错原因 #

该报错通常由以下情况导致：

• event['body']不存在或为None：比如Lambda触发源不是API Gateway POST请求，或测试事件未配置合法的body字段
• 请求body不是合法JSON格式：比如发送的是表单数据而非JSON，或JSON语法存在错误
• 未捕获解析异常导致函数直接崩溃

修复后代码 #

在lambda_handler中添加异常捕获与参数校验逻辑：

import json
import boto3
import os
import logging
import random

def generate_random_number():
    return random.randint(1, 1000000)

thingName = 'abc'

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    # 打印event结构，便于排查body问题
    logger.info(f"Received event: {json.dumps(event)}")
    
    # 处理JSON解析异常
    try:
        payload_data = json.loads(event['body']) if event.get('body') else {}
    except json.JSONDecodeError as e:
        logger.error(f"JSON解析失败: {str(e)}")
        return {
            'statusCode': 400,
            'headers': {
                'Access-Control-Allow-Headers': 'Content-Type',
                'Access-Control-Allow-Origin': '*',
                'Access-Control-Allow-Methods': 'OPTIONS,POST'
            },
            'body': json.dumps({'error': f"无效的JSON格式: {str(e)}"})
        }
    
    data = payload_data.get('inputs')
    if not data:
        logger.error("请求中缺少inputs字段")
        return {
            'statusCode': 400,
            'headers': {
                'Access-Control-Allow-Headers': 'Content-Type',
                'Access-Control-Allow-Origin': '*',
                'Access-Control-Allow-Methods': 'OPTIONS,POST'
            },
            'body': json.dumps({'error': "请求body中必须包含inputs字段"})
        }
    
    # 将IoT客户端初始化移到handler内部，避免冷启动资源浪费
    iot_client = boto3.client('iot', region_name=os.environ['AWS_REGION'])
    endpoint_response = iot_client.describe_endpoint(endpointType='iot:Data-ATS')
    endpoint_url = f"https://{endpoint_response['endpointAddress']}"
    iot_shadow_client = boto3.client('iot-data', region_name=os.environ['AWS_REGION'], endpoint_url=endpoint_url)
    
    payload_str = json.dumps(data)
    payload_bytes = payload_str.encode('utf-8')
    logger.info(payload_bytes)
    
    update_response = iot_shadow_client.update_thing_shadow(
        thingName=thingName,
        payload=payload_bytes
    )
    output = update_response.get('payload').read().decode("utf-8")
    logger.info(output)
    
    # 生成随机ID（用于后续DynamoDB条目主键）
    random_id = generate_random_number()
    logger.info(f"生成的随机ID: {random_id}")

    return {
        'statusCode': 200,
        'headers': {
            'Access-Control-Allow-Headers': 'Content-Type',
            'Access-Control-Allow-Origin': '*',
            'Access-Control-Allow-Methods': 'OPTIONS,POST'
        },
        'body': output
    }

二、无PutItem实现DynamoDB条目创建（结合随机数） #

若要求不能直接调用DynamoDB的put_item方法，可通过AWS IoT规则引擎实现自动写入：

步骤1：创建IoT规则 #

1. 进入AWS IoT控制台，导航到规则页面，点击「创建规则」
2. 规则查询语句：用SQL生成随机ID并捕获Shadow更新数据

   SELECT random() as id, current.state.reported as data FROM '$aws/things/abc/shadow/update/documents'
   

   • random()：生成随机数作为DynamoDB主键
   • $aws/things/abc/shadow/update/documents：监听目标设备Shadow的更新事件
3. 添加动作：选择「将消息写入DynamoDB表」
   • 选择目标DynamoDB表
   • 设置主键名称（如id），值来源选择id字段
   • 配置其他字段映射，将Shadow数据写入表中

步骤2：验证流程 #

当Lambda调用update_thing_shadow更新设备Shadow后，IoT规则会自动触发，将包含随机ID的Shadow数据写入DynamoDB表，全程无需在Lambda中调用PutItem。

额外说明 #

原代码中DynamoDB相关的循环逻辑完全无效，已移除。若后续允许直接操作DynamoDB，可添加以下代码（不符合「不能使用Put-Item」要求）：

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('你的表名')
# 写入数据，用随机ID作为主键
table.put_item(
    Item={
        'id': random_id,
        'shadow_data': json.loads(output)
    }
)

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