如何将BOTO3 Dynamo DB项目转换为Python中的常规字典

Question

在python中，当使用boto3从dynamo db检索项目时，获得了以下架构。

{
  "ACTIVE": {
    "BOOL": true
  },
  "CRC": {
    "N": "-1600155180"
  },
  "ID": {
    "S": "bewfv43843b"
  },
  "params": {
    "M": {
      "customer": {
        "S": "TEST"
      },
      "index": {
        "N": "1"
      }
    }
  },
  "THIS_STATUS": {
    "N": "10"
  },
  "TYPE": {
    "N": "22"
  }
}

同样，当插入或扫描时，词典必须以这种方式转换。我找不到可以照顾这种转换的包装纸。由于显然BOTO3不支持这一点，因此是否有比实施代码更好的选择？

Answer 1

为了了解如何解决这个问题，重要的是要认识到boto3具有两种基本操作模式：一种使用低级客户端API，一种使用更高级别的抽象喜欢桌子。问题中显示的数据结构是低级API消耗/生产的示例，AWS CLI和DynamoDB Web服务也使用。

回答您的问题 - 如果您可以专门与 table 这样的高级抽象工作，那么在使用boto3时，这对您来说会容易得多，正如评论所暗示的那样。然后，您可以避开整个问题 - python类型已访问和从低级数据格式进行了封储。

但是，在某些时候不可能专门使用这些高级构造。当处理附加到lambdas的DynamoDB流时，我特别遇到了这个问题。lambda的输入始终为低级格式，这种格式很难与IMO一起使用。

挖掘后，我发现boto3本身具有一些nifty功能，以进行转换。这些功能在前面提到的所有内部转换中隐含地使用。要直接使用它们，请导入Typedeserializer/ShoxeRializer类，并将它们与诸如So的DICT综合结合在一起：

import boto3
low_level_data = {
  "ACTIVE": {
    "BOOL": True
  },
  "CRC": {
    "N": "-1600155180"
  },
  "ID": {
    "S": "bewfv43843b"
  },
  "params": {
    "M": {
      "customer": {
        "S": "TEST"
      },
      "index": {
        "N": "1"
      }
    }
  },
  "THIS_STATUS": {
    "N": "10"
  },
  "TYPE": {
    "N": "22"
  }
}
# Lazy-eval the dynamodb attribute (boto3 is dynamic!)
boto3.resource('dynamodb')
# To go from low-level format to python
deserializer = boto3.dynamodb.types.TypeDeserializer()
python_data = {k: deserializer.deserialize(v) for k,v in low_level_data.items()}
# To go from python to low-level format
serializer = boto3.dynamodb.types.TypeSerializer()
low_level_copy = {k: serializer.serialize(v) for k,v in python_data.items()}
assert low_level_data == low_level_copy

Answer 2

您可以使用TypedeSerializer类

from boto3.dynamodb.types import TypeDeserializer
deserializer = TypeDeserializer()
document = { "ACTIVE": { "BOOL": True }, "CRC": { "N": "-1600155180" }, "ID": { "S": "bewfv43843b" }, "params": { "M": { "customer": { "S": "TEST" }, "index": { "N": "1" } } }, "THIS_STATUS": { "N": "10" }, "TYPE": { "N": "22" } }
deserialized_document = {k: deserializer.deserialize(v) for k, v in document.items()}
print(deserialized_document)

Answer 3

有一个python软件包，称为" dynamodb-json"这可以帮助您实现这一目标。DynamoDB-JSON util的工作原理与JSON负载和转储功能相同。我更喜欢使用它，因为它可以固有地转换小数对象。

您可以找到示例以及如何通过此链接进行安装-https：//pypi.org/project/dynamodb-json/

Answer 4

我下来写了一个自定义解决方案

它没有覆盖所有类型，但足以让我使用的类型。任何人都可以进一步发展的好起点

from re import compile as re_compile

class Serializer:
    re_number = re_compile(r"^-?d+?.?d*$")
    def serialize(self, data: any) -> dict:
        if isinstance(data, bool):  # booleans are a subtype of integers so place above int
            return {'BOOL': data}
        if isinstance(data, (int, float)):
            return {'N': str(data)}
        if isinstance(data, type(None)) or not data:  # place below int (0) and bool (False)
            # returns NULL for empty list, tuple, dict, set or string
            return {'NULL': True}
        if isinstance(data, (list, tuple)):
            return {'L': [self.serialize(v) for v in data]}
        if isinstance(data, set):
            if all([isinstance(v, str) for v in data]):
                return {'SS': data}
            if all([self.re_number.match(str(v)) for v in data]):
                return {'NS': [str(v) for v in data]}
        if isinstance(data, dict):
            return {'M': {k: self.serialize(v) for k, v in data.items()}}
        return {'S': str(data)}  # safety net to catch all others
    def deserialize(self, data: dict) -> dict:
        _out = {}
        if not data:
            return _out
        for k, v in data.items():
            if k in ('S', 'SS', 'BOOL'):
                return v
            if k == 'N':
                return float(v) if '.' in v else int(v)
            if k == 'NS':
                return [float(_v) if '.' in _v else int(_v) for _v in v]
            if k == 'M':
                return {_k: self.deserialize(_v) for _k, _v in v.items()}
            if k == 'L':
                return [self.deserialize(_v) for _v in v]
            if k == 'NULL':
                return None
            _out[k] = self.deserialize(v)
        return _out

用法

serialized = Serializer().serialize(input_dict)
print(serialized)
deserialized = Serializer().deserialize(serialized)
print(deserialized)

dynamodb(python(

dynamodb = boto3.client('dynamodb')
dynamodb.put_item(
    TableName=table_name,
    Item={
        'id': {'S': id},
        'data': Serializer().serialize(data)
    }
)
response = dynamodb.get_item(
    TableName=table_name,
    Key={
        'id': {'S': id}
    }
)
data = Serializer().deserialize(response['Item'])