如何解析SQS JSON消息适配后端？本地测试遇KeyError问题

核心问题 #

你遇到的KeyError本质是两个原因：

1. handler函数未正确处理SQS事件结构：没有提取消息体传递给解析函数，也未调用解析逻辑。
2. 测试消息格式不符合SQS实际规范：SQS消息的body是JSON字符串，而非直接的JSON对象，导致json.loads解析失败，后续键提取出错。

步骤1：修正测试消息格式 #

将测试消息中的body字段改为JSON字符串（转义内部双引号），如下：

{
  "Records": [
    {
      "messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
      "receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
      "body": "{\"severity\":\"CRITICAL\",\"alertTs\": \"timestamp\", \"data\":\"P-123\", \"resource\":\"resource\",\"resourceId\":\"riskybucket4testing\", \"policyType\": \"config\", \"policyName\": \"PolicyNameTest\", \"policyId\": \"4daa435b-fa46-457a-9359-6a4b4a43a442\", \"alertRuleName\":\"JTB-SQS Alerts\",\"alertRemediationImpact\":\"null\",\"accountName\":\"jtb75-002-pan\",\"hasFinding\":\"false\",\"resourceRegionId\":\"us-east-1\",\"alertRemediationCli\":\"null\",\"source\":\"Prisma Cloud\",\"cloudType\":\"aws\", \"accountId\": \"12345678\", \"alertId\":\"P-1000\"}",
      "attributes": {
        "ApproximateReceiveCount": "1",
        "SentTimestamp": "1545082649183",
        "SenderId": "AIDAIENQZJOLO23YVJ4VO",
        "ApproximateFirstReceiveTimestamp": "1545082649185"
      },
      "messageAttributes": {},
      "md5OfBody": "098f6bcd4621d373cade4e832627b4f6",
      "eventSource": "aws:sqs",
      "eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:my-queue",
      "awsRegion": "us-east-2"
    }
  ]
}

步骤2：修正代码逻辑 #

完善解析函数 #

补充成功解析后的返回语句，同时添加键存在性检查，避免KeyError：

import json

def parse_alert_message(sqs_message):
    try:
        alert = json.loads(sqs_message)

        if alert['alertId'] == 'P-0':
            return {'error': "Error", 'data': alert['alertId']}

        # 检查必填字段是否存在
        required_fields = ['policyType', 'policyId', 'accountName', 'accountId', 'resourceRegionId', 'resource', 'severity']
        missing_fields = [field for field in required_fields if field not in alert]
        if missing_fields:
            return {'error': f"缺失必填字段: {', '.join(missing_fields)}", 'data': None}

        parsed_alert = {
            'policy_type': alert['policyType'],
            'policy_Id': alert['policyId'],
            'account': {
                'name': alert['accountName'],
                'account_id': alert['accountId']
            },
            'region': alert['resourceRegionId'],
            'metadata': alert['resource'],
            'severity': alert['severity']
        }
        return parsed_alert

    except Exception as e:
        return {'error': str(e), 'data': None}

修正handler函数 #

正确遍历SQS消息记录，提取消息体并调用解析函数：

def handler(event, context):
    processed_results = []
    for record in event['Records']:
        # 提取每条消息的body
        message_body = record['body']
        # 解析消息
        result = parse_alert_message(message_body)
        processed_results.append(result)
    
    return {'处理结果': processed_results}

关键说明 #

• SQS事件的Records是数组，每条记录的body才是实际的业务消息内容。
• 实际SQS传递的body是JSON字符串，所以测试时必须用转义后的字符串格式，否则json.loads会抛出类型错误。
• 添加必填字段检查可以提前发现消息格式问题，避免后续KeyError。

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