如何在 Python 中将 boto3 Dynamo DB 项目转换为常规字典？

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 有两种基本操作模式：一种使用低级 Client API，另一种使用高级抽象，如 Table。问题中显示的数据结构是低级 API 使用/生成的示例，AWS CLI 和 dynamodb Web 服务也使用它。

回答你的问题 - 如果你可以在使用 boto3 时专门使用像 Table 这样的高级抽象，那么正如评论所建议的那样，事情对你来说会容易得多。然后你可以回避整个问题 - python 类型为你编组进出低级数据格式。

然而，有些时候不可能专门使用那些高级构造。在处理附加到 Lambda 的 DynamoDB 流时，我特别遇到了这个问题。 lambda 的输入始终采用低级格式，这种格式更难与 IMO 一起使用。

经过一些挖掘，我发现 boto3 本身有一些漂亮的功能隐藏起来进行转换。这些功能在前面提到的所有内部转换中都隐式使用。要直接使用它们，请导入 TypeDeserializer/TypeSerializer 类并将它们与 dict comprehensions 结合起来，如下所示：

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

有一个名为“dynamodb-json”的python包可以帮助你实现这一点。 dynamodb-json 实用程序的工作方式与 json 加载和转储功能相同。我更喜欢使用它，因为它本身负责转换 Decimal 对象。

您可以通过以下链接找到示例以及如何安装它 - https://pypi.org/project/dynamodb-json/

Answer 4

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

它没有涵盖所有类型，但足以满足我使用的那些。任何人进一步发展的良好起点，

from re import compile as re_compile
from typing import Optional, Union


class Serialize:
    re_number = re_compile(r"^-?\d+?\.?\d*$")

    def serialize_scalar(self, v):
        if isinstance(v, type(None)):
            return {'NULL': True}
        if isinstance(v, bool):  # put before int
            return {'BOOL': v}
        if isinstance(v, (int, float)):  # put after bool
            return {'N': str(v)}
        return {'S': str(v)}

    def serialize_list(self, _list):
        _out = []
        if not [n for n in _list if not self.re_number.match(str(n))]:
            return {'NS': [str(n) for n in _list]}
        if not [s for s in _list if not isinstance(s, str)]:
            return {'SS': _list}
        for item in _list:
            if isinstance(item, dict):
                _out.append(self.serialize_dict(item))
            elif isinstance(item, (list, tuple)):
                _out.append(self.serialize_list(item))
            else:
                _out.append(self.serialize_scalar(item))
        return {'L': _out}

    def serialize_dict(self, data):
        _out = {}
        for key in data:
            if isinstance(data[key], dict):
                _out[key] = self.serialize_dict(data[key])
            elif isinstance(data[key], list):
                _out[key] = self.serialize_list(data[key])
            else:
                _out[key] = self.serialize_scalar(data[key])
        return {'M': _out}

    def serialize(self, data: Optional[Union[dict, list, str, int, float, bool]]):
        if isinstance(data, dict):
            return self.serialize_dict(data)
        if isinstance(data, (list, tuple)):
            return self.serialize_list(data)
        return self.serialize_scalar(data)


class Deserialize:
    def deserialize(self, data: dict):
        for k, v in data.items():
            if k == 'N':  # number
                return float(v) if '.' in v else int(v)
            if k in ('S', 'SS', 'BOOL'):  # string, string sequence, boolean
                return v
            if k == 'NS':  # number sequence
                return [float(_v) if '.' in _v else int(_v) for _v in v]
            if k == 'M':  # dict
                _out = {}
                for _k, _v in v.items():
                    _out[_k] = self.deserialize(_v)
                return _out
            if k == 'L':  # list
                _out = []
                for _v in v:
                    _out.append(self.deserialize(_v))
                return _out
            if k == 'NULL':  # null
                return None