Skip to content

Commit

Permalink
\rate_limited_scan fix to handle consumed capacity (#235)
Browse files Browse the repository at this point in the history
  • Loading branch information
@anandswaminathan @danielhochman
anandswaminathan authored and danielhochman committed Feb 22, 2017
1 parent e5858ef commit 4dddef0
Show file tree
Hide file tree
Showing 3 changed files with 20 additions and 20 deletions.
2 changes: 1 addition & 1 deletion pynamodb/connection/base.py
View file
Expand Up @@ -1019,7 +1019,7 @@ def rate_limited_scan(self,
if not limit:
return

latest_scan_consumed_capacity = data.get(CONSUMED_CAPACITY)
latest_scan_consumed_capacity = data.get(CONSUMED_CAPACITY).get(CAPACITY_UNITS)
last_evaluated_key = data.get(LAST_EVALUATED_KEY, None)
consecutive_provision_throughput_exceeded_ex = 0
except ScanError as e:
Expand Down
36 changes: 18 additions & 18 deletions pynamodb/tests/test_base_connection.py
View file
Expand Up @@ -1485,7 +1485,7 @@ def verify_scan_call_args(call_args,
self.assertEqual(call_args[1]['segment'], segment)

with patch(SCAN_METHOD_TO_PATCH) as req:
req.return_value = {'Items': [], 'ConsumedCapacity': 10 }
req.return_value = {'Items': [], 'ConsumedCapacity': {'TableName': table_name, 'CapacityUnits': 10.0}}
resp = conn.rate_limited_scan(
table_name
)
Expand All @@ -1494,7 +1494,7 @@ def verify_scan_call_args(call_args,
verify_scan_call_args(req.call_args, table_name)

with patch(SCAN_METHOD_TO_PATCH) as req:
req.return_value = {'Items': [], 'ConsumedCapacity': 10 }
req.return_value = {'Items': [], 'ConsumedCapacity': {'TableName': table_name, 'CapacityUnits': 10.0}}
resp = conn.rate_limited_scan(
table_name,
limit=10,
Expand All @@ -1511,7 +1511,7 @@ def verify_scan_call_args(call_args,
limit=10)

with patch(SCAN_METHOD_TO_PATCH) as req:
req.return_value = {'Items': [], 'ConsumedCapacity': 10 }
req.return_value = {'Items': [], 'ConsumedCapacity': {'TableName': table_name, 'CapacityUnits': 10.0}}
scan_filter = {
'ForumName': {
'AttributeValueList': [
Expand Down Expand Up @@ -1551,7 +1551,7 @@ def verify_scan_call_args(call_args,
conditional_operator='AND')

with patch(SCAN_METHOD_TO_PATCH) as req:
req.return_value = {'Items': [], 'ConsumedCapacity': 10 }
req.return_value = {'Items': [], 'ConsumedCapacity': {'TableName': table_name, 'CapacityUnits': 10.0}}
resp = conn.rate_limited_scan(
table_name,
page_size=5,
Expand All @@ -1565,7 +1565,7 @@ def verify_scan_call_args(call_args,
limit=5)

with patch(SCAN_METHOD_TO_PATCH) as req:
req.return_value = {'Items': [], 'ConsumedCapacity': 10 }
req.return_value = {'Items': [], 'ConsumedCapacity': {'TableName': table_name, 'CapacityUnits': 10.0}}
resp = conn.rate_limited_scan(
table_name,
limit=10,
Expand Down Expand Up @@ -1767,8 +1767,8 @@ def test_ratelimited_scan_with_pagination_ends(self, scan_mock, time_mock):
c = Connection()
time_mock.side_effect = [1, 10, 20, 30, 40]
scan_mock.side_effect = [
{'Items': ['Item-1'], 'ConsumedCapacity': 1, 'LastEvaluatedKey': 'XX' },
{'Items': ['Item-2'], 'ConsumedCapacity': 1 }
{'Items': ['Item-1'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 1}, 'LastEvaluatedKey': 'XX' },
{'Items': ['Item-2'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 1}}
]
resp = c.rate_limited_scan('Table_1')
values = list(resp)
Expand All @@ -1787,7 +1787,7 @@ def test_ratelimited_scan_retries_on_throttling(self, api_mock, sleep_mock, time

api_mock.side_effect = [
VerboseClientError(botocore_expected_format, 'operation_name', {}),
{'Items': ['Item-1', 'Item-2'], 'ConsumedCapacity': 40 }
{'Items': ['Item-1', 'Item-2'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 40}}
]
resp = c.rate_limited_scan('Table_1')
values = list(resp)
Expand Down Expand Up @@ -1853,8 +1853,8 @@ def test_rate_limited_scan_retries_on_rate_unavailable(self, scan_mock, sleep_mo
sleep_mock.return_value = 1
time_mock.side_effect = [1, 4, 6, 12]
scan_mock.side_effect = [
{'Items': ['Item-1'], 'ConsumedCapacity': 80, 'LastEvaluatedKey': 'XX' },
{'Items': ['Item-2'], 'ConsumedCapacity': 41 }
{'Items': ['Item-1'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 80}, 'LastEvaluatedKey': 'XX' },
{'Items': ['Item-2'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 41}}
]
resp = c.rate_limited_scan('Table_1')
values = list(resp)
Expand All @@ -1873,8 +1873,8 @@ def test_rate_limited_scan_retries_on_rate_unavailable_within_s(self, scan_mock,
sleep_mock.return_value = 1
time_mock.side_effect = [1.0, 1.5, 4.0]
scan_mock.side_effect = [
{'Items': ['Item-1'], 'ConsumedCapacity': 10, 'LastEvaluatedKey': 'XX' },
{'Items': ['Item-2'], 'ConsumedCapacity': 11 }
{'Items': ['Item-1'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 10}, 'LastEvaluatedKey': 'XX' },
{'Items': ['Item-2'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 11}}
]
resp = c.rate_limited_scan('Table_1', read_capacity_to_consume_per_second=5)
values = list(resp)
Expand All @@ -1892,8 +1892,8 @@ def test_rate_limited_scan_retries_max_sleep(self, scan_mock, sleep_mock, time_m
sleep_mock.return_value = 1
time_mock.side_effect = [1.0, 1.5, 250, 350]
scan_mock.side_effect = [
{'Items': ['Item-1'], 'ConsumedCapacity': 1000, 'LastEvaluatedKey': 'XX' },
{'Items': ['Item-2'], 'ConsumedCapacity': 11 }
{'Items': ['Item-1'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 1000}, 'LastEvaluatedKey': 'XX' },
{'Items': ['Item-2'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 11}}
]
resp = c.rate_limited_scan(
'Table_1',
Expand All @@ -1915,8 +1915,8 @@ def test_rate_limited_scan_retries_min_sleep(self, scan_mock, sleep_mock, time_m
sleep_mock.return_value = 1
time_mock.side_effect = [1, 2, 3, 4]
scan_mock.side_effect = [
{'Items': ['Item-1'], 'ConsumedCapacity': 10, 'LastEvaluatedKey': 'XX' },
{'Items': ['Item-2'], 'ConsumedCapacity': 11 }
{'Items': ['Item-1'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 10}, 'LastEvaluatedKey': 'XX' },
{'Items': ['Item-2'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 11}}
]
resp = c.rate_limited_scan('Table_1', read_capacity_to_consume_per_second=8)
values = list(resp)
Expand All @@ -1934,8 +1934,8 @@ def test_rate_limited_scan_retries_timeout(self, scan_mock, sleep_mock, time_moc
sleep_mock.return_value = 1
time_mock.side_effect = [1, 20, 30, 40]
scan_mock.side_effect = [
{'Items': ['Item-1'], 'ConsumedCapacity': 1000, 'LastEvaluatedKey': 'XX' },
{'Items': ['Item-2'], 'ConsumedCapacity': 11 }
{'Items': ['Item-1'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 1000}, 'LastEvaluatedKey': 'XX' },
{'Items': ['Item-2'], 'ConsumedCapacity': {'TableName': 'Table_1', 'CapacityUnits': 11}}
]
resp = c.rate_limited_scan(
'Table_1',
Expand Down
2 changes: 1 addition & 1 deletion pynamodb/tests/test_model.py
View file
Expand Up @@ -1973,7 +1973,7 @@ def test_rate_limited_scan(self):
item = copy.copy(GET_MODEL_ITEM_DATA.get(ITEM))
item['user_id'] = {STRING_SHORT: 'id-{0}'.format(idx)}
items.append(item)
req.return_value = {'Items': items}
req.return_value = {'Items': items, 'ConsumedCapacity': {'TableName': 'UserModel', 'CapacityUnits': 10}}
scan_result = UserModel.rate_limited_scan(
user_id__contains='tux',
zip_code__null=False,
Expand Down

0 comments on commit 4dddef0

Please sign in to comment.