使用Python实现DynamoDB流的序列化和非序列化|类型验证程序没有帮助

x33g5p2x 于 4个月前发布在 Python

关注(0)|答案(1)|浏览(79)

感谢您对我和社区的所有帮助和支持。
尝试在互联网上抛出SO和一些文章，并指出要对dynamoDBItem进行验证，一旦可以使用TypeValidizer，它就可以轻松工作（它处理数据类型，但不是通用的k，v对）
但是有没有一个标准的库，使用它可以将ddb流事件转换为包括newImage在内的事件？

{
     "eventID":"1",
     "eventName":"INSERT",
     "eventVersion":"1.0",
     "eventSource":"aws:dynamodb",
     "awsRegion":"us-east-1",
     "dynamodb":{
        "Keys":{
           "Id":{
              "N":"101"
           }
        },
        "NewImage":{
           "Message":{
              "S":"New item!"
           },
           "Id":{
              "N":"101"
           }
        },
        "SequenceNumber":"111",
        "SizeBytes":26,
        "StreamViewType":"NEW_AND_OLD_IMAGES"
     },
     "eventSourceARN":"stream-ARN"
  }

字符串

python-3.x

来源：https://stackoverflow.com/questions/71919160/serialization-and-deserialization-for-dynamodb-stream-with-python-typedeserial

1条答案

按热度按时间

cidc1ykv1#

我使用了我在另一个question中找到的代码片段。它工作得很好。

from boto3.dynamodb.types import TypeDeserializer

serializer = TypeDeserializer()

def deserialize(data):
    if isinstance(data, list):
        return [deserialize(v) for v in data]

    if isinstance(data, dict):
        try:
            return serializer.deserialize(data)
        except TypeError:
            return {k: deserialize(v) for k, v in data.items()}
    else:
        return data

字符串
已验证您的请求：

{
  "eventID": "1",
  "eventName": "INSERT",
  "eventVersion": "1.0",
  "eventSource": "aws:dynamodb",
  "awsRegion": "us-east-1",
  "dynamodb": {
    "Keys": {
      "Id": "101"
    },
    "NewImage": {
      "Message": "New item!",
      "Id": "101"
    },
    "SequenceNumber": "111",
    "SizeBytes": 26,
    "StreamViewType": "NEW_AND_OLD_IMAGES"
  },
  "eventSourceARN": "stream-ARN"
}

型

赞(0）回复(0）举报 4个月前

我来回答

使用Python实现DynamoDB流的序列化和非序列化|类型验证程序没有帮助

1条答案

相关问题

热门标签

最新问答