Lambda函数测试正常，经API Gateway调用时出现502错误

我有一个向DynamoDB表写入数据的Lambda函数，通过Lambda测试事件运行完全正常，但通过API Gateway调用时抛出内部服务器错误。

Lambda代码 #

import boto3

class DBupload:

    def __init__(self):
        client = boto3.resource('dynamodb')
        self.table=client.Table('labels')
    
    def Create_data(self,event):
        response=self.table.put_item(
            Item={
                'instanceID' : event['id'],
                'imageID' : event['imageid'],
                'labels' : event['labels']
            }
        )
        return{
            'statusCode' : response['ResponseMetadata']['HTTPStatusCode'],
            'body' : 'Record ' + event['id'] + ' added'
        }

def lambda_handler(event, context):
    if event:
        instance =  DBupload()
        if event['tasktype']  == "create":
            create_result =  instance.Create_data(event['data'])
            return create_result
        else :
            return {
                'statusCode': '404',
                'body': 'Not found'
            }

配置步骤 #

• 创建REST API Gateway，添加资源和带代理的POST方法
• 启用IAM和API密钥后部署
• 将阶段添加到使用计划，再次部署确保生效

错误现象 #

• Postman发送与测试事件相同的请求时，返回内部服务器错误
• 使用API Gateway测试工具（绕过认证）调用，得到错误日志：

Lambda execution failed with status 200 due to customer function error: 'tasktype'.
Method completed with status: 502

Lambda测试事件 #

{
    "tasktype":"create",
    "data":{
       "id":"testinsanceid",
       "imageid":"testimageid",
       "labels":{
          "labeltype1":"labelname1",
          "labeltype2":"labelname2"
       }
    }
}

问题原因 #

核心问题是API Gateway的代理集成会修改传入Lambda的事件结构：

• Lambda测试事件是直接将tasktype、data作为顶层字段传入
• 而带代理的POST方法会把请求体包装在event的body字段中（通常是JSON字符串格式），Lambda直接读取event['tasktype']自然会找不到该键，抛出KeyError。

代理集成传入的Lambda事件大致结构如下：

{
  "resource": "/your-resource-path",
  "path": "/your-resource-path",
  "httpMethod": "POST",
  "headers": {...},
  "body": "{\"tasktype\":\"create\",\"data\":{...}}",
  "isBase64Encoded": false
}

解决方案 #

方案1：修改Lambda代码适配代理集成结构 #

调整lambda_handler，先解析body字段的JSON内容，同时兼容测试事件的结构：

import json
import boto3

class DBupload:
    def __init__(self):
        client = boto3.resource('dynamodb')
        self.table=client.Table('labels')
    
    def Create_data(self,event):
        response=self.table.put_item(
            Item={
                'instanceID' : event['id'],
                'imageID' : event['imageid'],
                'labels' : event['labels']
            }
        )
        return{
            'statusCode' : response['ResponseMetadata']['HTTPStatusCode'],
            'body' : 'Record ' + event['id'] + ' added'
        }

def lambda_handler(event, context):
    # 解析API Gateway代理传入的body，兼容测试事件
    if event.get('body'):
        try:
            event_body = json.loads(event['body'])
        except json.JSONDecodeError:
            return {
                'statusCode': 400,
                'body': json.dumps('Invalid JSON payload')
            }
    else:
        event_body = event
    
    if event_body:
        instance = DBupload()
        if event_body.get('tasktype') == "create":
            create_result = instance.Create_data(event_body['data'])
            # API Gateway要求返回的body为字符串，需序列化
            return {
                'statusCode': create_result['statusCode'],
                'body': json.dumps(create_result['body'])
            }
        else :
            return {
                'statusCode': 404,
                'body': json.dumps('Not found')
            }
    else:
        return {
            'statusCode': 400,
            'body': json.dumps('Empty request')
        }

注意：将原代码中字符串类型的'404'改为数字类型404，符合HTTP状态码规范。

方案2：修改API Gateway为非代理集成 #

如果不想改动Lambda代码，可以调整API Gateway的集成方式：

1. 在POST方法的集成设置中，取消勾选使用Lambda代理集成
2. 进入"集成请求"的"映射模板"，添加application/json模板，内容为：

$input.json('$')

这样API Gateway会直接将请求体作为Lambda的事件输入，结构与测试事件完全一致。

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