はじめに
DynamoDBから値を取得するときに、batch_get_itemの戻り値やDynamoDB Streamのイベントに格納されるデータの形式が、次のようにデータ型を含んだ形になっていて、そのままでは扱いづらくなっています。
{
'userId': {
'N': '12345'
},
'userName': {
'S': 'ユーザ1'
},
'mailAddresses': {
'L': [
{
'S': 'hoge@example.com'
},
{
'S': 'fuga@example.com'
}
]
}
}
数値も文字列になってしまっています。
この形式のデータを
{
'userId': 12345,
'userName': 'ユーザ1',
'mailAddresses': [
'hoge@example.com',
'fuga@example.com'
]
}
に直すコードが欲しかったので、書いてみました。
コード
remove_data_type の引数に、データ型を含むdictを渡すと、データ型に従って変換されたdictが返ります。
データ型毎に処理を書いているだけです。
一部、再帰が入っています。
import base64
def remove_data_type(record):
return {k: remove_one_data_type(record[k]) for k in record}
def remove_one_data_type(data_type_and_value_dict):
data_type, value = list(data_type_and_value_dict.items())[0]
if data_type == 'S' or data_type == 'BOOL':
return value
if data_type == 'N':
return float(value) if '.' in value else int(value)
if data_type == 'B':
return base64.b64decode(value)
if data_type == 'NS':
return set([float(v) if '.' in v else int(v) for v in value])
if data_type == 'SS':
return set(value)
if data_type == 'BS':
return set([base64.b64decode(v) for v in value])
if data_type == 'M':
return remove_data_type(value)
if data_type == 'L':
return [remove_one_data_type(v) for v in value]
if data_type == 'NULL':
return None
raise Exception('Invalid data type: ' + data_type)
次のように呼び出します。
dynamodb_record = {
'userId': {
'N': '12345'
},
'userName': {
'S': 'ユーザ1'
},
'mailAddresses': {
'L': [
{
'S': 'hoge@example.com'
},
{
'S': 'fuga@example.com'
}
]
}
}
converted_record = remove_data_type(dynamodb_record)
print(converted_record)
出力
{
'userId': 12345,
'userName': 'ユーザ1',
'mailAddresses': [
'hoge@example.com',
'fuga@example.com'
]
}