Python Lambda函数解析DynamoDB的JSON格式

new9mtju  于 2023-11-20  发布在  Python
关注(0)|答案(6)|浏览(154)

为dynamodb流调用的Python Lambda函数具有具有DynamoDB格式的JSON(包含JSON中的数据类型)。我想将DynamoDB JSON转换为标准JSON。PHP和nodejs具有Marshaler可以做到这一点。请让我知道Python是否有类似或其他选项。

DynamoDB_format = `{"feas":
    {"M": {
        "fea": {
            "L": [
                {
                    "M": {
                        "pre": {
                            "N": "1"
                        },
                        "Li": {
                            "N": "1"
                        },
                        "Fa": {
                            "N": "0"
                        },
                        "Mo": {
                            "N": "1"
                        },
                        "Ti": {
                            "S": "20160618184156529"
                        },
                        "Fr": {
                            "N": "4088682"
                        }
                    }
                }
                ]
            }   
        }
    }
}`

字符串

zc0qhyus

zc0qhyus1#

更新:现在有一个库:https://pypi.org/project/dynamodb-json/

下面是indiangolfer's answer的一个改进版本。虽然@indiangolfer的解决方案可以解决这个问题,但这个改进版本可能对偶然发现这个帖子的人更有用。

def unmarshal_dynamodb_json(node):
    data = dict({})
    data['M'] = node
    return _unmarshal_value(data)

def _unmarshal_value(node):
    if type(node) is not dict:
        return node

    for key, value in node.items():
        # S – String - return string
        # N – Number - return int or float (if includes '.')
        # B – Binary - not handled
        # BOOL – Boolean - return Bool
        # NULL – Null - return None
        # M – Map - return a dict
        # L – List - return a list
        # SS – String Set - not handled
        # NN – Number Set - not handled
        # BB – Binary Set - not handled
        key = key.lower()
        if key == 'bool':
            return value
        if key == 'null':
            return None
        if key == 's':
            return value
        if key == 'n':
            if '.' in str(value):
                return float(value)
            return int(value)
        if key in ['m', 'l']:
            if key == 'm':
                data = {}
                for key1, value1 in value.items():
                    if key1.lower() == 'l':
                        data = [_unmarshal_value(n) for n in value1]
                    else:
                        if type(value1) is not dict:
                            return _unmarshal_value(value)
                        data[key1] = _unmarshal_value(value1)
                return data
            data = []
            for item in value:
                data.append(_unmarshal_value(item))
            return data

字符串
它在以下方面得到了改进:

  • 处理更多的data types,包括以前没有正确处理的列表
  • 手柄和按键

编辑:修复递归对象错误

wpx232ag

wpx232ag2#

我在野外找不到任何东西。所以,我决定将dynamodb json的PHP实现移植到here发布的标准json。我在处理DynamoDB流的python lambda函数中测试了这一点。如果有更好的方法,请告诉我。
(PS:这不是PHP Marshaler的完整端口)
问题中的JSON被转换为:

{  
   "feas":{  
      "fea":[  
         {  
            "pre":"1",
            "Mo":"1",
            "Ti":"20160618184156529",
            "Fa":"0",
            "Li":"1",
            "Fr":"4088682"
         }
      ]
   }
}

个字符

nkkqxpd9

nkkqxpd93#

为了轻松地转换DynamoDB JSON,我建议使用boto3 dynamodb类型序列化器和序列化器。

import boto3
from boto3.dynamodb.types import TypeSerializer, TypeDeserializer
ts= TypeSerializer()
td = TypeDeserializer()

data= {"id": "5000"}
serialized_data= ts.serialize(data)
print(serialized_data)
#{'M': {'id': {'S': '5000'}}}
deserialized_data= td.deserialize(serialized_data)
print(deserialized_data)
#{'id': '5000'}

字符串
更多详情check out the boto3.dynamodb.types classes .

o2gm4chl

o2gm4chl4#

根据this blog,以下似乎是最简单的解决方案:

from boto3.dynamodb.types import TypeDeserializer, TypeSerializer

def unmarshall(dynamo_obj: dict) -> dict:
    """Convert a DynamoDB dict into a standard dict."""
    deserializer = TypeDeserializer()
    return {k: deserializer.deserialize(v) for k, v in dynamo_obj.items()}

def marshall(python_obj: dict) -> dict:
    """Convert a standard dict into a DynamoDB ."""
    serializer = TypeSerializer()
    return {k: serializer.serialize(v) for k, v in python_obj.items()}

字符串

vm0i2vca

vm0i2vca5#

这对我很有效。对@vekerdyb的回答做了小修改

def _unmarshalValue(ddbValue):                                                       
    for key, value in ddbValue.items():                                              
        if key.lower() == "s":                                                       
            return value                                                             
        elif key.lower() == "n":                                                     
            return int(value)                                                        
        elif key.lower() == "bool":                                                  
            return value                                                             
        elif key.lower() == "m":                                                     
            data = {}                                                                
            for mKey, mValue in value.items():                                       
                data[mKey] = _unmarshalValue(mValue)                                 
            return data                                                              
        elif key.lower() == "l":                                                     
            data = []                                                                
            for item in value:                                                       
                data.append(_unmarshalValue(item))                                   
            return data                                                              
                                                                                 
                                                                                 
def unmarshalDynamoDBJson(ddbItem):                                                                                                                                                                                                                                                                                                                    
    result = {}                                                                      
    for key, value in ddbItem.items():                                               
        result[key] = _unmarshalValue(value)                                         
                                                                                 
    return result

字符串

i86rm4rw

i86rm4rw6#

import json import boto3 import base64
输出= []
def lambda_handler(event,context):print(event)for record in event 'records']:payload = base64.b64decode(record 'data']).decode('utf-8')print('payload:',payload)

row_w_newline = payload + "\n"
    print('row_w_newline type:', type(row_w_newline))
    row_w_newline = base64.b64encode(row_w_newline.encode('utf-8'))
    
    output_record = {
        'recordId': record['recordId'],
        'result': 'Ok',
        'data': row_w_newline
    }
    output.append(output_record)

print('Processed {} records.'.format(len(event['records'])))

return {'records': output}

字符串

相关问题