How to use the pynamodb.expressions.operand.Path function in pynamodb

To help you get started, we’ve selected a few pynamodb examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github pynamodb / PynamoDB / tests / test_base_connection.py View on Github external
},
                    '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'
github pynamodb / PynamoDB / tests / test_base_connection.py View on Github external
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,
github pynamodb / PynamoDB / tests / test_base_connection.py View on Github external
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'
                    }
github pynamodb / PynamoDB / tests / test_base_connection.py View on Github external
'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': {
github pynamodb / PynamoDB / tests / test_base_connection.py View on Github external
'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,
github pynamodb / PynamoDB / tests / test_base_connection.py View on Github external
"""
        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,
github pynamodb / PynamoDB / pynamodb / attributes.py View on Github external
def __ge__(self, other):
        return Path(self).__ge__(other)
github pynamodb / PynamoDB / pynamodb / attributes.py View on Github external
def __lt__(self, other):
        return Path(self).__lt__(other)
github pynamodb / PynamoDB / pynamodb / attributes.py View on Github external
def exists(self):
        return Path(self).exists()
github pynamodb / PynamoDB / pynamodb / attributes.py View on Github external
def set(self, value):
        return Path(self).set(value)