Repository: pynamodb/PynamoDB

Issue:
Error while _handle_binary_attributes when UNPROCESSED_KEYS in batch_get request
Hi,
I encounter some problems whenever I exceed the provisionned throughput (if it does not exceed everything works well). Instead of retrying to request unprocessed items, it raises an error while trying to handle binary attributes.

This occurs when trying to process batch_get requests.

```
/Users/leonardbinet/anaconda/envs/api_transilien/lib/python3.6/site-packages/pynamodb/connection/base.py in dispatch(self, operation_name, operation_kwargs)
    262         self._log_debug(operation_name, operation_kwargs)
    263
--> 264         data = self._make_api_call(operation_name, operation_kwargs)
    265
    266         if data and CONSUMED_CAPACITY in data:

/Users/leonardbinet/anaconda/envs/api_transilien/lib/python3.6/site-packages/pynamod/connection/base.py in _make_api_call(self, operation_name, operation_kwargs)
    355                         continue
    356
--> 357             return self._handle_binary_attributes(data)
    358
    359     @staticmethod

/Users/leonardbinet/anaconda/envs/api_transilien/lib/python3.6/site-packages/pynamodb/connection/base.py in _handle_binary_attributes(data)
    378             for item_list in six.itervalues(data[UNPROCESSED_KEYS]):
    379                 for item in item_list:
--> 380                     for attr in six.itervalues(item):
    381                         _convert_binary(attr)
    382         if UNPROCESSED_ITEMS in data:

/Users/leonardbinet/anaconda/envs/api_transilien/lib/python3.6/site-packages/six.py in itervalues(d, **kw)
    576
    577     def itervalues(d, **kw):
--> 578         return iter(d.values(**kw))
    579
    580     def iteritems(d, **kw):

AttributeError: 'str' object has no attribute 'values'
```

Here is more detail about objects when investigating with ipython:
```
ipdb> d
'Keys'

ipdb> item
'Keys'

ipdb> item_list.keys()
dict_keys(['Keys'])

ipdb> item_list
{'Keys': [{'station_id': {'S': '8739322'}, 'day_train_num': {'S': '20170401_496702'}}, {'station_id': {'S':'8738149'}, 'day_train_num': {'S': '20170401_497601'}}, 
...
...
 {'station_id': {'S': '8727608'}, 'day_train_num': {'S': '20170401_496538'}}, {'station_id': {'S': '8738237'}, 'day_train_num': {'S': '20170401_496704'}}
]}

ipdb> data.keys()
dict_keys(['ConsumedCapacity', 'Responses', 'UnprocessedKeys'])

ipdb> data["UnprocessedKeys"]
{'real_departures_2': {'Keys': [{'station_id': {'S': '8739322'}, 'day_train_num': {'S': '20170401_496702'}}, {'station_id': {'S': '8738149'}, 'day_train_num': {'S': '20170401_497601'}}, {'station_id': {'S': '8711386'}, 'day_train_num': {'S': '20170401_499001'}}, {'station_id': {'S': '8738237'}, 'day_train_num': {'S': '20170401_496703'}},
.... 
```

Obviously the problem is that item_list variable is not a list, but a dictionary containing the item_list as value under key "Keys": 
`{"Keys": item_list}`

And that initially: 
`data["UnprocessedKeys"]`is structured as follow:
`{"table_name":{"Keys": item_list}}`
instead of `{"Keys": item_list}`

I think that we only need to add a line in code (line 377 in pynamodb/connection/base.py): 
Instead of:
```
if UNPROCESSED_KEYS in data:
            for item_list in six.itervalues(data[UNPROCESSED_KEYS]):
                for item in item_list:
                    for attr in six.itervalues(item):
                        _convert_binary(attr)
``` 
insert another loop:

```
if UNPROCESSED_KEYS in data:
            for table_unprocessed_keys in six.itervalues(data[UNPROCESSED_KEYS]):
                for item_list in six.itervalues(table_unprocessed_keys):
                    for item in item_list:
                        for attr in six.itervalues(item):
                            _convert_binary(attr)
``` 

Discussion:

Patch:
diff --git a/docs/conditional.rst b/docs/conditional.rst
index 12521a7..bcac16c 100644
--- a/docs/conditional.rst
+++ b/docs/conditional.rst
@@ -1,3 +1,5 @@
+.. _conditional_operations:
+
 Conditional Operations
 ======================
 
diff --git a/docs/local.rst b/docs/local.rst
index 2349727..982d09e 100644
--- a/docs/local.rst
+++ b/docs/local.rst
@@ -15,9 +15,9 @@ that your server is listening on.
 .. note::
 
     If you are using DynamoDB Local and also use ``rate_limited_scan`` on your models, you must also
-    set ``allow_rate_limited_scan_without_consumed_capacity`` to ``True`` in the
-    `settings file <settings.rst#allow_rate_limited_scan_without_consumed_capacity>`_ (dynalite does not require
-    this step because it implements returning of consumed capacity in responses, which is used by ``rate_limited_scan``).
+    set ``allow_rate_limited_scan_without_consumed_capacity`` to ``True`` in :ref:`settings`
+    (dynalite does not require this step because it implements returning of consumed capacity in
+    responses, which is used by ``rate_limited_scan``).
 
 .. note::
 
diff --git a/docs/quickstart.rst b/docs/quickstart.rst
index 3d46d71..1405c46 100644
--- a/docs/quickstart.rst
+++ b/docs/quickstart.rst
@@ -32,7 +32,7 @@ PynamoDB allows you to create the table:
 
     >>> UserModel.create_table(read_capacity_units=1, write_capacity_units=1)
 
-Now you can create a user:
+Now you can create a user in local memory:
 
     >>> user = UserModel('test@example.com', first_name='Samuel', last_name='Adams')
     dynamodb-user<test@example.com>
@@ -62,6 +62,49 @@ Ready to delete the user?
 
     >>> user.delete()
 
+Changing items
+^^^^^^^^^^^^^^
+
+Changing existing items in the database can be done using either
+`update()` or `save()`. There are important differences between the
+two.
+
+Use of `save()` looks like this::
+
+    user = UserModel.get('test@example.com')
+    user.first_name = 'Robert'
+    user.save()
+
+Use of `update()` (in its simplest form) looks like this::
+
+    user = UserModel.get('test@example.com')
+    user.update(
+      actions=[
+        UserModel.first_name.set('Robert')
+      ]
+    )
+
+`save()` will entirely replace an object (it internally uses `PutItem
+<http://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_PutItem.html>`_). As
+a consequence, even if you modify only one attribute prior to calling
+`save()`, the entire object is re-written. Any modifications done to
+the same user by other processes will be lost, even if made to other
+attributues that you did not change. To avoid this, use `update()` to
+perform more fine grained updates or see the
+:ref:`conditional_operations` for how to avoid race conditions
+entirely.
+
+Additionally, PynamoDB ignores attributes it does not know about when
+reading an object from the database. As a result, if the item in
+DynamoDB contains attributes not declared in your model, `save()` will
+cause those attributes to be deleted.
+
+In particular, performing a rolling upgrade of your application after
+having added an attribute is an example of such a situation. To avoid
+data loss, either avoid using `save()` or perform a multi-step update
+with the first step is to upgrade to a version that merely declares
+the attribute on the model without ever setting it to any value.
+
 Querying
 ^^^^^^^^
 
diff --git a/docs/settings.rst b/docs/settings.rst
index f61e00a..3b9288a 100644
--- a/docs/settings.rst
+++ b/docs/settings.rst
@@ -1,3 +1,5 @@
+.. _settings:
+
 Settings
 ========
 
@@ -59,7 +61,9 @@ Default: ``False``
 
 If ``True``, ``rate_limited_scan()`` will proceed silently (without
 rate limiting) if the DynamoDB server does not return consumed
-capacity information in responses.
+capacity information in responses. If ``False``, scans will fail
+should the server not return consumed capacity information in an
+effort to prevent unintentional capacity usage..
 
 Overriding settings
 ~~~~~~~~~~~~~~~~~~~
diff --git a/docs/tutorial.rst b/docs/tutorial.rst
index dd03776..6b2db70 100644
--- a/docs/tutorial.rst
+++ b/docs/tutorial.rst
@@ -102,7 +102,7 @@ Here is an example that specifies both the ``host`` and the ``region`` to use:
             table_name = 'Thread'
             # Specifies the region
             region = 'us-west-1'
-            # Specifies the hostname
+            # Optional: Specify the hostname only if it needs to be changed from the default AWS setting
             host = 'http://localhost'
             # Specifies the write capacity
             write_capacity_units = 10
diff --git a/pynamodb/connection/base.py b/pynamodb/connection/base.py
index ae163f5..ac92ae0 100644
--- a/pynamodb/connection/base.py
+++ b/pynamodb/connection/base.py
@@ -420,8 +420,8 @@ class Connection(object):
             for attr in six.itervalues(data[LAST_EVALUATED_KEY]):
                 _convert_binary(attr)
         if UNPROCESSED_KEYS in data:
-            for item_list in six.itervalues(data[UNPROCESSED_KEYS]):
-                for item in item_list:
+            for table_data in six.itervalues(data[UNPROCESSED_KEYS]):
+                for item in table_data[KEYS]:
                     for attr in six.itervalues(item):
                         _convert_binary(attr)
         if UNPROCESSED_ITEMS in data:
diff --git a/pynamodb/models.py b/pynamodb/models.py
index 92dda13..af4b42b 100644
--- a/pynamodb/models.py
+++ b/pynamodb/models.py
@@ -61,9 +61,14 @@ class BatchWrite(ModelContextManager):
     """
     def save(self, put_item):
         """
-        This adds `put_item` to the list of pending writes to be performed.
-        Additionally, the a BatchWriteItem will be performed if the length of items
-        reaches 25.
+        This adds `put_item` to the list of pending operations to be performed.
+
+        If the list currently contains 25 items, which is the DynamoDB imposed
+        limit on a BatchWriteItem call, one of two things will happen. If auto_commit
+        is True, a BatchWriteItem operation will be sent with the already pending
+        writes after which put_item is appended to the (now empty) list. If auto_commit
+        is False, ValueError is raised to indicate additional items cannot be accepted
+        due to the DynamoDB imposed limit.
 
         :param put_item: Should be an instance of a `Model` to be written
         """
@@ -76,8 +81,14 @@ class BatchWrite(ModelContextManager):
 
     def delete(self, del_item):
         """
-        This adds `del_item` to the list of pending deletes to be performed.
-        If the list of items reaches 25, a BatchWriteItem will be called.
+        This adds `del_item` to the list of pending operations to be performed.
+
+        If the list currently contains 25 items, which is the DynamoDB imposed
+        limit on a BatchWriteItem call, one of two things will happen. If auto_commit
+        is True, a BatchWriteItem operation will be sent with the already pending
+        operations after which put_item is appended to the (now empty) list. If auto_commit
+        is False, ValueError is raised to indicate additional items cannot be accepted
+        due to the DynamoDB imposed limit.
 
         :param del_item: Should be an instance of a `Model` to be deleted
         """
@@ -294,9 +305,13 @@ class Model(AttributeContainer):
     @classmethod
     def batch_write(cls, auto_commit=True):
         """
-        Returns a context manager for a batch operation'
+        Returns a BatchWrite context manager for a batch operation.
 
-        :param auto_commit: Commits writes automatically if `True`
+        :param auto_commit: If true, the context manager will commit writes incrementally
+                            as items are written to as necessary to honor item count limits
+                            in the DynamoDB API (see BatchWrite). Regardless of the value
+                            passed here, changes automatically commit on context exit
+                            (whether successful or not).
         """
         return BatchWrite(cls, auto_commit=auto_commit)
 
@@ -455,7 +470,8 @@ class Model(AttributeContainer):
     def get(cls,
             hash_key,
             range_key=None,
-            consistent_read=False):
+            consistent_read=False,
+            attributes_to_get=None):
         """
         Returns a single object using the provided keys
 
@@ -466,7 +482,8 @@ class Model(AttributeContainer):
         data = cls._get_connection().get_item(
             hash_key,
             range_key=range_key,
-            consistent_read=consistent_read
+            consistent_read=consistent_read,
+            attributes_to_get=attributes_to_get
         )
         if data:
             item_data = data.get(ITEM)


Test patch:
diff --git a/pynamodb/tests/test_base_connection.py b/pynamodb/tests/test_base_connection.py
index 5d36fad..4095d6c 100644
--- a/pynamodb/tests/test_base_connection.py
+++ b/pynamodb/tests/test_base_connection.py
@@ -2586,6 +2586,44 @@ class ConnectionTestCase(TestCase):
             _assert=True
         )
 
+    def test_handle_binary_attributes_for_unprocessed_keys(self):
+        binary_blob = six.b('\x00\xFF\x00\xFF')
+        unprocessed_keys = {
+            'UnprocessedKeys': {
+                'MyTable': {
+                    'AttributesToGet': ['ForumName'],
+                    'Keys': [
+                        {
+                            'ForumName': {'S': 'FooForum'},
+                            'Subject': {'B': base64.b64encode(binary_blob).decode(DEFAULT_ENCODING)}
+                        },
+                        {
+                            'ForumName': {'S': 'FooForum'},
+                            'Subject': {'S': 'thread-1'}
+                        }
+                    ],
+                    'ConsistentRead': False
+                },
+                'MyOtherTable': {
+                    'AttributesToGet': ['ForumName'],
+                    'Keys': [
+                        {
+                            'ForumName': {'S': 'FooForum'},
+                            'Subject': {'B': base64.b64encode(binary_blob).decode(DEFAULT_ENCODING)}
+                        },
+                        {
+                            'ForumName': {'S': 'FooForum'},
+                            'Subject': {'S': 'thread-1'}
+                        }
+                    ],
+                    'ConsistentRead': False
+                }
+            }
+        }
+        data = Connection._handle_binary_attributes(unprocessed_keys)
+        self.assertEqual(data['UnprocessedKeys']['MyTable']['Keys'][0]['Subject']['B'], binary_blob)
+        self.assertEqual(data['UnprocessedKeys']['MyOtherTable']['Keys'][0]['Subject']['B'], binary_blob)
+
     def test_get_expected_map(self):
         conn = Connection(self.region)
         with patch(PATCH_METHOD) as req: