Added support for dynamo update_item on nested dicts (#1345)

This commit is contained in:
Terry Cain 2017-11-17 08:49:59 +00:00 committed by GitHub
parent cdb7305dac
commit 2609f6cd3a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 92 additions and 2 deletions

View File

@ -149,9 +149,38 @@ class Item(BaseModel):
key = key.strip()
value = value.strip()
if value in expression_attribute_values:
self.attrs[key] = DynamoType(expression_attribute_values[value])
value = DynamoType(expression_attribute_values[value])
else:
self.attrs[key] = DynamoType({"S": value})
value = DynamoType({"S": value})
if '.' not in key:
self.attrs[key] = value
else:
# Handle nested dict updates
key_parts = key.split('.')
attr = key_parts.pop(0)
if attr not in self.attrs:
raise ValueError()
last_val = self.attrs[attr].value
for key_part in key_parts:
# Hack but it'll do, traverses into a dict
if list(last_val.keys())[0] == 'M':
last_val = last_val['M']
if key_part not in last_val:
raise ValueError()
last_val = last_val[key_part]
# We have reference to a nested object but we cant just assign to it
current_type = list(last_val.keys())[0]
if current_type == value.type:
last_val[current_type] = value.value
else:
last_val[value.type] = value.value
del last_val[current_type]
elif action == 'ADD':
key, value = value.split(" ", 1)
key = key.strip()

View File

@ -1006,3 +1006,64 @@ def test_query_missing_expr_names():
resp['Count'].should.equal(1)
resp['Items'][0]['client']['S'].should.equal('test2')
# https://github.com/spulec/moto/issues/1342
@mock_dynamodb2
def test_update_item_on_map():
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
# Create the DynamoDB table.
dynamodb.create_table(
TableName='users',
KeySchema=[
{
'AttributeName': 'forum_name',
'KeyType': 'HASH'
},
{
'AttributeName': 'subject',
'KeyType': 'RANGE'
},
],
AttributeDefinitions=[
{
'AttributeName': 'forum_name',
'AttributeType': 'S'
},
{
'AttributeName': 'subject',
'AttributeType': 'S'
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5
}
)
table = dynamodb.Table('users')
table.put_item(Item={
'forum_name': 'the-key',
'subject': '123',
'body': {'nested': {'data': 'test'}},
})
resp = table.scan()
resp['Items'][0]['body'].should.equal({'nested': {'data': 'test'}})
table.update_item(Key={
'forum_name': 'the-key',
'subject': '123'
},
UpdateExpression='SET body.#nested.#data = :tb',
ExpressionAttributeNames={
'#nested': 'nested',
'#data': 'data'
},
ExpressionAttributeValues={
':tb': 'new_value'
})
resp = table.scan()
resp['Items'][0]['body'].should.equal({'nested': {'data': 'new_value'}})