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

Question

在Python中，使用boto3从DynamoDB检索项目时，会获得以下类似于模式的结果。

{
  "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不支持此功能，所以是否有比实现此功能更好的替代方法？

同时在插入或者扫描时，需要按照这种方式转换字典。目前我还没有找到能够处理这种转换的封装器。既然boto3似乎不支持这个功能，那么除了编写代码之外，是否有更好的替代方案呢？

- manelmc

---

响应语法已经是 dict。无需转换。 boto3 文档显示它是一个 dict 对象 http://boto3.readthedocs.io/en/latest/reference/services/dynamodb.html#DynamoDB.Client.get_item - mootmoot

是的，它是一个字典对象，但类型必须明确。这就是为什么我称之为转换的原因。 - manelmc

也许你的问题标题与你的需求不符，请更改标题并澄清你需要什么。请记住，没有针对 NoSQL 结果的“标准”解析器。你需要处理每个元素数据定义。 - mootmoot

boto3зљ„Table资源可以为您完成此操作。文档 - Jordon Phillips

谢谢@JordonPhillips，这正是我在寻找的，如果我的问题有误导之处，请见谅。 - manelmc

也许这个短语应该是“一个简单的字典”... - MikeW

Answer 1

为了理解如何解决这个问题，重要的是要认识到boto3有两种基本操作模式:一种使用低级别的Client API，另一种使用高级抽象，比如Table。问题中展示的数据结构是低级API消耗/产生的示例，该API也被AWS CLI和dynamodb web服务所使用。

回答您的问题 - 如果您在使用boto3时只使用高级抽象（如Table），那么事情会变得更容易，正如评论所建议的那样。然后，您可以绕过整个问题，因为Python类型已经为您编排为低级数据格式。

但是，在某些情况下，仅使用这些高级构造不可能。我特别在处理连接到Lambdas的DynamoDB流时遇到了这个问题。Lambda的输入始终处于低级格式中，而且在我看来，该格式很难使用。

经过一番调查，我发现boto3本身具有一些巧妙的隐藏功能，可以进行转换。这些功能在所有先前提到的内部转换中都是隐式使用的。要直接使用它们，请导入TypeDeserializer/TypeSerializer类，然后与字典推导一起组合使用，如下所示：

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

- killthrush

---

更好的做法是，同时兼容Python2版本： python_data = deserializer.deserialize({'M': low_level_data}) - aaa90210

请注意，在使用boto3==1.9.79版本时，我需要以不同的方式导入反序列化程序：from boto3.dynamodb.types import TypeDeserializer。模块源代码显示，反序列化程序已经不像最初由 @killthrush 解释的那样公开暴露出来了。 - Eric Platon

这个工作非常好！我试图将一个dynamodb表复制到另一个表中，我不得不使用低级API +高级API来进行批量写入。这救了我一命。谢谢！ - amaurs

请注意，如果您的"low_level"来自于json.loads，则此代码不支持'B'（二进制类型），因为数据是utf-8字符串，而需要的是base64字节。我必须预处理并查找“B”，或者仅针对此情况在deserializer._deserialize_b上打猴子补丁以进行b64解码。 - Pierre-Francoys Brousseau

这对于“L”类型不起作用。需要递归反序列化它们。 - Joe

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)

- Fellipe

---

它对我有效。谢谢 ;) - Son Lam

必须使用boto3.resource('dynamodb')，否则会出现AttributeError: module 'boto3' has no attribute 'dynamodb'。 - Jorge Tovar

Answer 3

有一个名为 "dynamodb-json" 的 Python 包可以帮助您实现此目的。dynamodb-json 实用程序与 json loads 和 dumps 函数相同。我更喜欢使用它，因为它会自动处理转换 Decimal 对象。

您可以通过访问此链接找到示例和安装说明 - https://pypi.org/project/dynamodb-json/

- aamir23

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

- Christian