DynamoDB Binary类型数据反序列化报TypeError问题排查

对DynamoDB数据执行反序列化操作时，使用如下代码：

import json
import base64
import decimal
from boto3.dynamodb.types import TypeDeserializer
from base64 import b64encode

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()}

## 转换逻辑代码
base64.b64encode((json.dumps(from_dynamodb_to_json(payload["dynamodb"]["NewImage"]), cls=DecimalEncoder) + '\n').encode("utf-8")).decode("utf-8")

代码运行时抛出类型错误，提示传入值必须为<class 'bytearray'>、<class 'bytes'>类型，错误栈如下：

[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]))

触发报错的payload样例数据如下：

{'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'}

报错出现在boto3的TypeDeserializer处理二进制（B）类型字段的环节。

DynamoDB Streams传递的事件数据中，二进制（B）类型字段的值是经过base64编码的字符串，但boto3原生TypeDeserializer的二进制反序列化逻辑，要求输入值必须是bytes/bytearray类型的原始字节对象，直接传入base64字符串会触发类型校验失败。

自定义继承TypeDeserializer的反序列化类，重写二进制类型的处理逻辑：先将事件中拿到的base64字符串解码为bytes类型，再交给原生反序列化逻辑处理即可。
修复后的完整代码如下：

import json
import base64
import decimal
from boto3.dynamodb.types import TypeDeserializer, Binary
from base64 import b64encode, b64decode

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)

# 自定义适配DynamoDB Streams数据格式的反序列化器
class StreamTypeDeserializer(TypeDeserializer):
    def _deserialize_b(self, value):
        # 兼容Streams传入的base64字符串格式二进制值
        if isinstance(value, str):
            value = b64decode(value)
        return Binary(value)

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

# 原有转换逻辑无需修改
base64.b64encode((json.dumps(from_dynamodb_to_json(payload["dynamodb"]["NewImage"]), cls=DecimalEncoder) + '\n').encode("utf-8")).decode("utf-8")

修复后二进制字段会被正确反序列化为boto3的Binary对象，后续DecimalEncoder会自动将其转为base64字符串，可正常完成JSON序列化与后续base64编码操作，不会再抛出类型错误。

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