如何通过HTTP-API(v2)+Lambda+DynamoDB规范返回JSON响应？

核心疑问 #

• 从REST API(v1)迁移到HTTP API(v2)，原v1中的Model功能在v2里的推荐替代方案是什么？
• 除了返回前手动处理数据库响应外，有没有更规整的方式返回数据？

具体问题与场景 #

1. 保留关键字字段操作报错 #

调用UpdateItem操作时发生错误(ValidationException)：无效的UpdateExpression：属性名称是保留关键字"owner"
（涉及字段：owner、name）

2. Decimal类型JSON序列化失败 #

Decimal类型对象无法JSON序列化

3. 全表查询代码报错 #

尝试用query请求所有行时，触发以下错误：

调用Query操作时发生错误(ValidationException)：请求中必须指定KeyConditions或KeyConditionExpression参数

现有代码要么过于繁琐（手动映射字段+自定义Decimal编码器），要么无法实现全表查询需求。

解决方案 #

一、HTTP API(v2)替代v1 Model功能的方案 #

HTTP API(v2)没有直接对应v1的Model功能，推荐两种实用替代方式：

• API Gateway请求/响应模板：在HTTP API的集成配置中，通过Velocity模板（VTL）直接转换DynamoDB响应格式，无需在Lambda中手动处理字段映射、类型转换。
• Lambda层复用逻辑：将字段映射、Decimal序列化等通用逻辑封装到Lambda层，所有相关函数直接引入复用，避免重复代码。

二、解决保留关键字报错 #

DynamoDB的保留关键字必须通过ExpressionAttributeNames别名引用，适用于所有操作类型：

UpdateItem示例

response = table.update_item(
    Key={'id': '123'},
    UpdateExpression='SET #own = :val1, #nam = :val2',
    ExpressionAttributeNames={
        '#own': 'owner',
        '#nam': 'name'
    },
    ExpressionAttributeValues={
        ':val1': 'new_owner',
        ':val2': 'new_name'
    }
)

Scan/Query示例（包含保留关键字字段）

response = table.scan(
    ProjectionExpression='#own, #nam, col1',
    ExpressionAttributeNames={
        '#own': 'owner',
        '#nam': 'name'
    }
)

三、简化Decimal序列化 #

无需自定义DecimalEncoder，两种简洁处理方式：

1. 全局配置类型转换
   利用boto3的TypeDeserializer自动转换DynamoDB返回的类型：

import boto3
from boto3.dynamodb.types import TypeDeserializer

def deserialize_item(item):
    deserializer = TypeDeserializer()
    return {k: deserializer.deserialize(v) for k, v in item.items()}

# 使用：
response = table.scan(...)
items = [deserialize_item(item) for item in response['Items']]

2. 序列化时直接处理
   在json.dumps中通过default参数一次性处理Decimal：

import json
from decimal import Decimal

serialized_data = json.dumps(items, default=lambda x: float(x) if isinstance(x, Decimal) else x)

四、实现全表查询 #

query操作必须指定主键条件，全表查询需使用scan，优化后简洁代码示例：

import json
import boto3
from decimal import Decimal

def lambda_handler(event, context):
    try:
        dynamodb = boto3.resource('dynamodb')
        table = dynamodb.Table("mytable")
        
        # 全表扫描，若有保留关键字需添加ExpressionAttributeNames
        response = table.scan(
            ProjectionExpression='col1, col2, col3, col4, col5, col6'
            # 示例：包含保留关键字owner时的配置
            # ExpressionAttributeNames={'#own': 'owner'},
            # ProjectionExpression='#own, col1, col2'
        )
        
        # 处理Decimal并序列化响应
        serialized_items = json.dumps(
            response['Items'],
            default=lambda x: float(x) if isinstance(x, Decimal) else x
        )
        
        return {
            "statusCode": 200,
            "headers": {"Content-Type": "application/json"},
            "body": serialized_items
        }

    except Exception as e:
        return {
            "statusCode": 500,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps({"error": str(e)})
        }

若表数据量较大，需循环调用scan并传入ExclusiveStartKey（值为上一次响应的LastEvaluatedKey）实现分页。

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