Use attr.get_value(value) when deserialize (#450)

change update & update_item to use attr.get_value(value) in deserilazation
pynamodb · Mar 1, 2018 · 8ab34d0 · 8ab34d0
1 parent 9f94288
commit 8ab34d0
Show file tree

Hide file tree

Showing 3 changed files with 139 additions and 3 deletions.
diff --git a/pynamodb/attributes.py b/pynamodb/attributes.py
@@ -468,7 +468,7 @@ def get_value(self, value):
         # previously, BOOL was serialized as N
         value_to_deserialize = super(BooleanAttribute, self).get_value(value)
         if value_to_deserialize is None:
-            value_to_deserialize = json.loads(value.get(NUMBER_SHORT, 0))
+            value_to_deserialize = json.loads(value.get(NUMBER_SHORT, '0'))
         return value_to_deserialize
 
 

diff --git a/pynamodb/models.py b/pynamodb/models.py
@@ -389,7 +389,7 @@ def update_item(self, attribute, value=None, action=None, condition=None, condit
             attr_name = self._dynamo_to_python_attr(name)
             attr = self._get_attributes().get(attr_name)
             if attr:
-                setattr(self, attr_name, attr.deserialize(value.get(ATTR_TYPE_MAP[attr.attr_type])))
+                setattr(self, attr_name, attr.deserialize(attr.get_value(value)))
         return data
 
     def update(self, attributes=None, actions=None, condition=None, conditional_operator=None, **expected_values):
@@ -441,7 +441,7 @@ def update(self, attributes=None, actions=None, condition=None, conditional_oper
             attr_name = self._dynamo_to_python_attr(name)
             attr = self._get_attributes().get(attr_name)
             if attr:
-                setattr(self, attr_name, attr.deserialize(value.get(ATTR_TYPE_MAP[attr.attr_type])))
+                setattr(self, attr_name, attr.deserialize(attr.get_value(value)))
         return data
 
     def save(self, condition=None, conditional_operator=None, **expected_values):

diff --git a/pynamodb/tests/integration/test_migration.py b/pynamodb/tests/integration/test_migration.py
@@ -154,3 +154,139 @@ class Meta:
     BAModel.create_table(read_capacity_units=1, write_capacity_units=1)
     BAModel('pkey', flag=True).save()
     assert (0, 0) == migrate_boolean_attributes(BAModel, ['flag'], allow_rate_limited_scan_without_consumed_capacity=True)
+
+
+@pytest.mark.parametrize("flag_value",[True, False, None])
+@pytest.mark.ddblocal
+def test_legacy_boolean_attribute_deserialization_in_update(ddb_url, flag_value):
+    class BAModel(Model):
+        class Meta:
+            table_name = 'lba_deserialization_test'
+            host = ddb_url
+        id = UnicodeAttribute(hash_key=True)
+        flag = BooleanAttribute(null=True)
+        value = UnicodeAttribute(null=True)
+
+    class LBAModel(Model):
+        class Meta:
+            table_name = 'lba_deserialization_test'
+            host = ddb_url
+        id = UnicodeAttribute(hash_key=True)
+        flag = LegacyBooleanAttribute(null=True)
+        value = UnicodeAttribute(null=True)
+
+    BAModel.create_table(read_capacity_units=1, write_capacity_units=1)
+
+    # Create objects with a BooleanAttribute flag
+    BAModel('pkey', flag=flag_value, value = 'value').save()
+
+    # Check we are able to read the flag with LegacyBooleanAttribute
+    assert flag_value == LBAModel.get('pkey').flag
+
+    # Update a value in the model causing LegacyBooleanAttribute to be deserialized
+    LBAModel.get('pkey').update(actions=[LBAModel.value.set('new value')])
+
+    # Check we are able to read the flag with LegacyBooleanAttribute
+    assert flag_value == LBAModel.get('pkey').flag
+
+
+@pytest.mark.parametrize("flag_value",[True, False, None])
+@pytest.mark.ddblocal
+def test_legacy_boolean_attribute_deserialization_in_update_item(ddb_url, flag_value):
+    class BAModel(Model):
+        class Meta:
+            table_name = 'lba_deserialization_test'
+            host = ddb_url
+        id = UnicodeAttribute(hash_key=True)
+        flag = BooleanAttribute(null=True)
+        value = UnicodeAttribute(null=True)
+
+    class LBAModel(Model):
+        class Meta:
+            table_name = 'lba_deserialization_test'
+            host = ddb_url
+        id = UnicodeAttribute(hash_key=True)
+        flag = LegacyBooleanAttribute(null=True)
+        value = UnicodeAttribute(null=True)
+
+    BAModel.create_table(read_capacity_units=1, write_capacity_units=1)
+
+    # Create objects with a BooleanAttribute flag
+    BAModel('pkey', flag=flag_value, value = 'value').save()
+
+    # Check we are able to read the flag with LegacyBooleanAttribute
+    assert flag_value == LBAModel.get('pkey').flag
+
+    # Update a value in the model causing LegacyBooleanAttribute to be deserialized
+    LBAModel.get('pkey').update_item('value', 'new value', 'PUT')
+
+    # Check we are able to read the flag with LegacyBooleanAttribute
+    assert flag_value == LBAModel.get('pkey').flag
+
+
+@pytest.mark.parametrize("flag_value",[True, False, None])
+@pytest.mark.ddblocal
+def test_boolean_attribute_deserialization_in_update(ddb_url, flag_value):
+    class BAModel(Model):
+        class Meta:
+            table_name = 'ba_deserialization_test'
+            host = ddb_url
+        id = UnicodeAttribute(hash_key=True)
+        flag = BooleanAttribute(null=True)
+        value = UnicodeAttribute(null=True)
+
+    class LBAModel(Model):
+        class Meta:
+            table_name = 'ba_deserialization_test'
+            host = ddb_url
+        id = UnicodeAttribute(hash_key=True)
+        flag = LegacyBooleanAttribute(null=True)
+        value = UnicodeAttribute(null=True)
+
+    LBAModel.create_table(read_capacity_units=1, write_capacity_units=1)
+
+    # Create an object with a LegacyBooleanAttribute flag
+    LBAModel('pkey', flag=flag_value, value = 'value').save()
+
+    # Check we are able to read the flag with BooleanAttribute
+    assert flag_value == BAModel.get('pkey').flag
+
+    # Update a value in the model causing BooleanAttribute to be deserialized
+    BAModel.get('pkey').update(actions=[BAModel.value.set('new value')])
+
+    # Check we are able to read the flag with BooleanAttribute
+    assert flag_value == BAModel.get('pkey').flag
+
+
+@pytest.mark.parametrize("flag_value",[True, False, None])
+@pytest.mark.ddblocal
+def test_boolean_attribute_deserialization_in_update_item(ddb_url, flag_value):
+    class BAModel(Model):
+        class Meta:
+            table_name = 'ba_deserialization_test'
+            host = ddb_url
+        id = UnicodeAttribute(hash_key=True)
+        flag = BooleanAttribute(null=True)
+        value = UnicodeAttribute(null=True)
+
+    class LBAModel(Model):
+        class Meta:
+            table_name = 'ba_deserialization_test'
+            host = ddb_url
+        id = UnicodeAttribute(hash_key=True)
+        flag = LegacyBooleanAttribute(null=True)
+        value = UnicodeAttribute(null=True)
+
+    LBAModel.create_table(read_capacity_units=1, write_capacity_units=1)
+
+    # Create an object with a LegacyBooleanAttribute flag
+    LBAModel('pkey', flag=flag_value, value = 'value').save()
+
+    # Check we are able to read the flag with BooleanAttribute
+    assert flag_value == BAModel.get('pkey').flag
+
+    # Update a value in the model causing BooleanAttribute to be deserialized
+    BAModel.get('pkey').update_item('value', 'new value', 'PUT')
+
+    # Check we are able to read the flag with BooleanAttribute
+    assert flag_value == BAModel.get('pkey').flag