DynamoDB值得二进制数据

Question

即时通讯使用以下代码对DynamoDB数据进行估算。

class DecimalEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, set):
            return list(o)
        if isinstance(o, decimal.Decimal):
            if o % 1 > 0:
                return float(o)
            else:
                return int(o)
        if isinstance(o, bytes):
            return b64encode(o).decode()        
        return super(DecimalEncoder, self).default(o)


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

## code to convert
base64.b64encode((json.dumps(from_dynamodb_to_json(payload["dynamodb"]["NewImage"]), cls=DecimalEncoder) + '\n').encode("utf-8")).decode("utf-8")

但是它给出以下错误。

[ERROR] TypeError: Value must be of the following types: <class 'bytearray'>, <class 'bytes'>.
Traceback (most recent call last):
  File "/var/task/lambda_function.py", line 60, in lambda_handler
    'data': base64.b64encode((json.dumps(from_dynamodb_to_json(payload["dynamodb"]["NewImage"]), cls=DecimalEncoder) + '\n').encode("utf-8")).decode("utf-8")
  File "/var/task/lambda_function.py", line 33, in from_dynamodb_to_json
    return {k: TypeDeserializer().deserialize(value=v) for k, v in item.items()}
  File "/var/task/lambda_function.py", line 33, in <dictcomp>
    return {k: TypeDeserializer().deserialize(value=v) for k, v in item.items()}
  File "/var/runtime/boto3/dynamodb/types.py", line 271, in deserialize
    return deserializer(value[dynamodb_type])
  File "/var/runtime/boto3/dynamodb/types.py", line 286, in _deserialize_b
    return Binary(value)
  File "/var/runtime/boto3/dynamodb/types.py", line 51, in __init__
    ', '.join([str(t) for t in BINARY_TYPES]))

有效载荷中的示例数据

{'awsRegion': 'ap-south-1', 'eventID': 'd926b17c-33a7-4d05-b936-7f5d9cd36a52', 'eventName': 'INSERT', 'userIdentity': None, 'recordFormat': 'application/json', 'tableName': 'dev_bhuvi_de', 'dynamodb': {'ApproximateCreationDateTime': 1656405526550, 'Keys': {'id': {'N': '261'}}, 'NewImage': {'id': {'N': '261'}, 'NewValue': {'B': 'YkdsdWRYaG9hVzUwTG1OdmJRbz0='}}, 'SizeBytes': 32}, 'eventSource': 'aws:dynamodb'}

Answer 1

当尝试从python lambda中的dynamoDB流解析数据时，我遇到了同一问题。

问题是，当发生DynamoDB更改时，发送给Lambda处理程序的事件字典具有所有二进制类型base64 base64编码。您的解决方案朝着正确的方向前进。

请参阅下面添加的有关如何解决此问题的__ deserialize（）方法：

import base64
from boto3.dynamodb.types import TypeDeserializer

lambda_event = {
    'awsRegion': 'us-east-0',
    'eventID': 'd926b17c-33a7-4d05-b936-7f5d9cd36a52',
    'eventName': 'INSERT',
    'userIdentity': None,
    'recordFormat': 'application/json',
    'tableName': 'my_table_name',
    'dynamodb': {
        'ApproximateCreationDateTime': 1656405526550,
        'Keys': {'id': {'N': '261'}},
        'NewImage': {'id': {'N': '261'}, 'NewValue': {'B': 'YkdsdWRYaG9hVzUwTG1OdmJRbz0='}},
        'SizeBytes': 32
    },
    'eventSource': 'aws:dynamodb'
}

def from_dynamodb_to_json(item):
    return {k: __deserialize(value=v) for k, v in item.items()}

def __deserialize(value):
    dynamodb_type = list(value.keys())[0]

    # DynamoDb stream base64-encodes all binary types, must base64 decode first
    if dynamodb_type == "B" and isinstance(value["B"], str):
        value = {"B": base64.b64decode(value["B"])}

    return TypeDeserializer().deserialize(value)

result = from_dynamodb_to_json(lambda_event["dynamodb"]["NewImage"])
print(result)