Add consistent_read parameter to Model.scan (#311)

Author: Lita Cho

pynamodb/connection/base.py

@@ -982,7 +982,8 @@ def rate_limited_scan(self,
                           read_capacity_to_consume_per_second=10,
                           allow_rate_limited_scan_without_consumed_capacity=None,
                           max_sleep_between_retry=10,
-                          max_consecutive_exceptions=10):
+                          max_consecutive_exceptions=10,
+                          consistent_read=None):
         """
         Performs a rate limited scan on the table. The API uses the scan API to fetch items from
         DynamoDB. The rate_limited_scan uses the 'ConsumedCapacity' value returned from DynamoDB to
@@ -1007,6 +1008,7 @@ def rate_limited_scan(self,
             throttling/rate limit scenarios
         :param max_consecutive_exceptions: Max number of consecutive ProvisionedThroughputExceededException
             exception for scan to exit
+        :param consistent_read: enable consistent read
         """
         read_capacity_to_consume_per_ms = float(read_capacity_to_consume_per_second) / 1000
         if allow_rate_limited_scan_without_consumed_capacity is None:
@@ -1038,7 +1040,8 @@ def rate_limited_scan(self,
                         return_consumed_capacity=TOTAL,
                         scan_filter=scan_filter,
                         segment=segment,
-                        total_segments=total_segments
+                        total_segments=total_segments,
+                        consistent_read=consistent_read
                     )
                     for item in data.get(ITEMS):
                         yield item
@@ -1120,7 +1123,8 @@ def scan(self,
              return_consumed_capacity=None,
              exclusive_start_key=None,
              segment=None,
-             total_segments=None):
+             total_segments=None,
+             consistent_read=None):
         """
         Performs the scan operation
         """
@@ -1154,6 +1158,8 @@ def scan(self,
                     operation_kwargs[SCAN_FILTER][key][ATTR_VALUE_LIST] = values
             if conditional_operator:
                 operation_kwargs.update(self.get_conditional_operator(conditional_operator))
+        if consistent_read:
+            operation_kwargs[CONSISTENT_READ] = consistent_read
         try:
             return self.dispatch(SCAN, operation_kwargs)
         except BOTOCORE_EXCEPTIONS as e:

pynamodb/connection/table.py

@@ -144,7 +144,8 @@ def rate_limited_scan(self,
              read_capacity_to_consume_per_second=None,
              allow_rate_limited_scan_without_consumed_capacity=None,
              max_sleep_between_retry=None,
-             max_consecutive_exceptions=None):
+             max_consecutive_exceptions=None,
+             consistent_read=None):
         """
         Performs the scan operation with rate limited
         """
@@ -162,7 +163,8 @@ def rate_limited_scan(self,
             read_capacity_to_consume_per_second=read_capacity_to_consume_per_second,
             allow_rate_limited_scan_without_consumed_capacity=allow_rate_limited_scan_without_consumed_capacity,
             max_sleep_between_retry=max_sleep_between_retry,
-            max_consecutive_exceptions=max_consecutive_exceptions)
+            max_consecutive_exceptions=max_consecutive_exceptions,
+            consistent_read=consistent_read)
 
     def scan(self,
              attributes_to_get=None,
@@ -172,7 +174,8 @@ def scan(self,
              return_consumed_capacity=None,
              segment=None,
              total_segments=None,
-             exclusive_start_key=None):
+             exclusive_start_key=None,
+             consistent_read=None):
         """
         Performs the scan operation
         """
@@ -185,7 +188,8 @@ def scan(self,
             return_consumed_capacity=return_consumed_capacity,
             segment=segment,
             total_segments=total_segments,
-            exclusive_start_key=exclusive_start_key)
+            exclusive_start_key=exclusive_start_key,
+            consistent_read=consistent_read)
 
     def query(self,
               hash_key,

pynamodb/models.py

@@ -675,6 +675,7 @@ def rate_limited_scan(cls,
             allow_rate_limited_scan_without_consumed_capacity=None,
             max_sleep_between_retry=10,
             max_consecutive_exceptions=30,
+            consistent_read=None,
             **filters):
         """
         Scans the items in the table at a definite rate.
@@ -698,6 +699,7 @@ def rate_limited_scan(cls,
             throttling/rate limit scenarios
         :param max_consecutive_exceptions: Max number of consecutive provision throughput exceeded
             exceptions for scan to exit
+        :param consistent_read: If True, a consistent read is performed
         """
 
         cls._conditional_operator_check(conditional_operator)
@@ -723,6 +725,7 @@ def rate_limited_scan(cls,
             allow_rate_limited_scan_without_consumed_capacity=allow_rate_limited_scan_without_consumed_capacity,
             max_sleep_between_retry=max_sleep_between_retry,
             max_consecutive_exceptions=max_consecutive_exceptions,
+            consistent_read=consistent_read,
         )
 
         for item in scan_result:
@@ -736,6 +739,7 @@ def scan(cls,
              conditional_operator=None,
              last_evaluated_key=None,
              page_size=None,
+             consistent_read=None,
              **filters):
         """
         Iterates through all items in the table
@@ -747,6 +751,7 @@ def scan(cls,
         :param last_evaluated_key: If set, provides the starting point for scan.
         :param page_size: Page size of the scan to DynamoDB
         :param filters: A list of item filters
+        :param consistent_read: If True, a consistent read is performed
         """
         cls._conditional_operator_check(conditional_operator)
         key_filter, scan_filter = cls._build_filters(
@@ -765,7 +770,8 @@ def scan(cls,
             limit=page_size,
             scan_filter=key_filter,
             total_segments=total_segments,
-            conditional_operator=conditional_operator
+            conditional_operator=conditional_operator,
+            consistent_read=consistent_read
         )
         log.debug("Fetching first scan page")
         last_evaluated_key = data.get(LAST_EVALUATED_KEY, None)

pynamodb/tests/test_base_connection.py

@@ -1606,6 +1606,23 @@ def test_scan(self):
             req.return_value = DESCRIBE_TABLE_DATA
             conn.describe_table(table_name)
 
+        with patch(PATCH_METHOD) as req:
+            req.return_value = {}
+            conn.scan(
+                table_name,
+                segment=0,
+                total_segments=22,
+                consistent_read=True
+            )
+            params = {
+                'ReturnConsumedCapacity': 'TOTAL',
+                'TableName': table_name,
+                'Segment': 0,
+                'TotalSegments': 22,
+                'ConsistentRead': True
+            }
+            self.assertDictEqual(req.call_args[0][1], params)
+
         with patch(PATCH_METHOD) as req:
             req.return_value = {}
             conn.scan(

pynamodb/tests/test_model.py

@@ -1906,6 +1906,16 @@ def fake_scan(*args):
             self.assertEquals(req.mock_calls[0][1][1]['Limit'], 4)
             self.assertEqual(count, 4)
 
+        with patch(PATCH_METHOD, new=mock_scan) as req:
+            count = 0
+            for item in UserModel.scan(limit=4, consistent_read=True):
+                count += 1
+                self.assertIsNotNone(item)
+            self.assertEqual(len(req.mock_calls), 2)
+            self.assertEquals(req.mock_calls[1][1][1]['Limit'], 4)
+            self.assertEquals(req.mock_calls[1][1][1]['ConsistentRead'], True)
+            self.assertEqual(count, 4)
+
         with patch(PATCH_METHOD) as req:
             items = []
             for idx in range(10):
@@ -1997,7 +2007,8 @@ def test_rate_limited_scan(self):
                 allow_rate_limited_scan_without_consumed_capacity=False,
                 max_sleep_between_retry=4,
                 max_consecutive_exceptions=22,
-                attributes_to_get=['X1', 'X2']
+                attributes_to_get=['X1', 'X2'],
+                consistent_read=True
             )
             self.assertEqual(1, len(list(result)))
             self.assertEqual('UserModel', req.call_args[0][0])
@@ -2014,7 +2025,8 @@ def test_rate_limited_scan(self):
                 'read_capacity_to_consume_per_second': 33,
                 'allow_rate_limited_scan_without_consumed_capacity': False,
                 'max_sleep_between_retry': 4,
-                'max_consecutive_exceptions': 22
+                'max_consecutive_exceptions': 22,
+                'consistent_read': True
             }
             self.assertEqual(params, req.call_args[1])
 

pynamodb/tests/test_table_connection.py

@@ -482,7 +482,8 @@ def test_rate_limited_scan(self):
                 read_capacity_to_consume_per_second=12,
                 allow_rate_limited_scan_without_consumed_capacity=False,
                 max_sleep_between_retry=3,
-                max_consecutive_exceptions=7
+                max_consecutive_exceptions=7,
+                consistent_read=True
             )
             self.assertEqual(self.test_table_name, req.call_args[0][0])
             params = {
@@ -498,6 +499,7 @@ def test_rate_limited_scan(self):
                 'read_capacity_to_consume_per_second': 12,
                 'allow_rate_limited_scan_without_consumed_capacity': False,
                 'max_sleep_between_retry': 3,
-                'max_consecutive_exceptions': 7
+                'max_consecutive_exceptions': 7,
+                'consistent_read': True
             }
             self.assertEqual(params, req.call_args[1])

requirements.txt

@@ -1,2 +1,2 @@
-botocore==1.1.2
+botocore==1.2.0
 six==1.9.0

setup.py

@@ -13,7 +13,7 @@
 
 install_requires = [
     'six',
-    'botocore>=1.0.0',
+    'botocore>=1.2.0',
     'python-dateutil>=2.1,<3.0.0',
 ]