Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
},
'Subject': {
'S': 'item1-range'
}
}
}
self.assertEqual(req.call_args[0][1], params)
with patch(PATCH_METHOD) as req:
req.return_value = {}
conn.put_item(
self.test_table_name,
'item1-hash',
range_key='item1-range',
attributes={'foo': {'S': 'bar'}},
condition=(Path('ForumName') == 'item1-hash')
)
params = {
'TableName': self.test_table_name,
'ConditionExpression': '#0 = :0',
'ExpressionAttributeNames': {
'#0': 'ForumName'
},
'ExpressionAttributeValues': {
':0': {
'S': 'item1-hash'
}
},
'ReturnConsumedCapacity': 'TOTAL',
'Item': {
'ForumName': {
'S': 'item1-hash'
def test_query(self):
"""
Connection.query
"""
conn = Connection()
table_name = 'Thread'
with patch(PATCH_METHOD) as req:
req.return_value = DESCRIBE_TABLE_DATA
conn.describe_table(table_name)
self.assertRaises(
ValueError,
conn.query,
table_name,
"FooForum",
Path('NotRangeKey').startswith('thread'),
Path('Foo') == 'Bar',
return_consumed_capacity='TOTAL'
)
self.assertRaises(
ValueError,
conn.query,
table_name,
"FooForum",
Path('Subject') != 'thread', # invalid sort key condition
return_consumed_capacity='TOTAL'
)
self.assertRaises(
ValueError,
conn.query,
with patch(PATCH_METHOD) as req:
req.return_value = {}
conn.scan(
table_name,
)
params = {
'ReturnConsumedCapacity': 'TOTAL',
'TableName': table_name
}
self.assertEqual(req.call_args[0][1], params)
with patch(PATCH_METHOD) as req:
req.return_value = {}
conn.scan(
table_name,
Path('ForumName').startswith('Foo') & Path('Subject').contains('Foo')
)
params = {
'ReturnConsumedCapacity': 'TOTAL',
'TableName': table_name,
'FilterExpression': '(begins_with (#0, :0) AND contains (#1, :1))',
'ExpressionAttributeNames': {
'#0': 'ForumName',
'#1': 'Subject'
},
'ExpressionAttributeValues': {
':0': {
'S': 'Foo'
},
':1': {
'S': 'Foo'
}
'Subject': {
'S': 'foo-range-key'
}
},
'TableName': self.test_table_name
}
self.assertEqual(req.call_args[0][1], params)
with patch(PATCH_METHOD) as req:
req.return_value = {}
conn.put_item(
self.test_table_name,
'item1-hash',
range_key='item1-range',
attributes={'foo': {'S': 'bar'}},
condition=(Path('Forum').does_not_exist() & (Path('Subject') == 'Foo'))
)
params = {
'ReturnConsumedCapacity': 'TOTAL',
'TableName': self.test_table_name,
'ConditionExpression': '(attribute_not_exists (#0) AND #1 = :0)',
'ExpressionAttributeNames': {
'#0': 'Forum',
'#1': 'Subject'
},
'ExpressionAttributeValues': {
':0': {
'S': 'Foo'
}
},
'Item': {
'ForumName': {
'S': 'How do I update multiple items?'
}
},
'TableName': self.test_table_name,
'ReturnItemCollectionMetrics': 'SIZE',
'ReturnConsumedCapacity': 'TOTAL'
}
self.assertEqual(req.call_args[0][1], params)
with patch(PATCH_METHOD) as req:
req.return_value = {}
conn.delete_item(
self.test_table_name,
"Amazon DynamoDB",
"How do I update multiple items?",
condition=Path('ForumName').does_not_exist(),
return_item_collection_metrics='SIZE'
)
params = {
'Key': {
'ForumName': {
'S': 'Amazon DynamoDB'
},
'Subject': {
'S': 'How do I update multiple items?'
}
},
'ConditionExpression': 'attribute_not_exists (#0)',
'ExpressionAttributeNames': {
'#0': 'ForumName'
},
'TableName': self.test_table_name,
"""
Connection.query
"""
conn = Connection()
table_name = 'Thread'
with patch(PATCH_METHOD) as req:
req.return_value = DESCRIBE_TABLE_DATA
conn.describe_table(table_name)
self.assertRaises(
ValueError,
conn.query,
table_name,
"FooForum",
Path('NotRangeKey').startswith('thread'),
Path('Foo') == 'Bar',
return_consumed_capacity='TOTAL'
)
self.assertRaises(
ValueError,
conn.query,
table_name,
"FooForum",
Path('Subject') != 'thread', # invalid sort key condition
return_consumed_capacity='TOTAL'
)
self.assertRaises(
ValueError,
conn.query,
table_name,
def __ge__(self, other):
return Path(self).__ge__(other)
def __lt__(self, other):
return Path(self).__lt__(other)
def exists(self):
return Path(self).exists()
def set(self, value):
return Path(self).set(value)