KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use dynamodb_client method in localstack

Best Python code snippet using localstack_python

server_monitor_context.py

Source:server_monitor_context.py

1"""2Module providing access to the Server Monitor data3"""4import boto35from botocore.exceptions import ClientError6import time7class ServerMonitorContext:8    """9    Context used to access the server monitor data in the DB10    """11    def __init__(self):12        """13        Initializes a new instance of the L{ServerMonitorContext} class 14        """15        self.table_name = 'servers'16        self.dynamodb_client = boto3.client('dynamodb', region_name='eu-central-1', endpoint_url="http://192.168.2.159:8000")17        try:18            self.dynamodb_client.describe_table(19                TableName = self.table_name)20        except ClientError as ce:21            if ce.response['Error']['Code'] == 'ResourceNotFoundException':22                self._initializeDb()23            else:24                raise25    26    def get_server(self, serverId):27        """28        Gets a server by its serverId29        """30        server = self.dynamodb_client.get_item(31            TableName=self.table_name,32            Key={33                'server_id':{34                    'S': serverId35                }36            }37        )38        ret = None39        if 'Item' in server:            40            ret = self._unbox_record(server['Item'])41        42        return ret43    44    def list_servers(self):45        """46        Gets a list of the servers in the DB47        """48        servers = self.dynamodb_client.scan(49            TableName=self.table_name)50        unboxed_servers = []51        for server in servers['Items']:52            unboxed_servers.append(self._unbox_record(server))53        return unboxed_servers54    def delete_server(self, serverId):55        """56        Deletes a server from the DB57        """58        response = self.dynamodb_client.delete_item(59            TableName=self.table_name,60            Key={61                'server_id':{62                    'S': serverId63                },64            },65            ReturnValues='ALL_OLD'66        )67        68        ret = False69        if 'Attributes' in response:70            ret = True71        return ret72    def add_server(self, serverData):73        """74        Adds a server to the DB75        """76        server = self._box_record(serverData)77        ret = True78        try:79            self.dynamodb_client.put_item(80                TableName=self.table_name,81                Item=server,82                ConditionExpression='attribute_not_exists(server_id)'83            )84        except ClientError as err:85            if err.response['Error']['Code'] == 'ConditionalCheckFailedException':86                ret = False87            else:88                raise89        return ret90    def update_server(self, serverData):91        """92        Updates the given server in the DB93        """94        server = self._box_record(serverData)95        ret = True96        try:97            self.dynamodb_client.put_item(98                TableName=self.table_name,99                Item=server,100                ConditionExpression='attribute_exists(server_id)'101            )102        except ClientError as err:103            if err.response['Error']['Code'] == 'ConditionalCheckFailedException':104                ret = False105            else:106                raise107                108        return ret109    def _box_record(self, record):110        """111        Encapsulates a record as DynamoDB is expecting it112        """113        for key in record.keys():114            if isinstance(record[key], int) or isinstance(record[key], float):115                record[key] = {'N': str(record[key])}116            else :117                record[key] = {'S': str(record[key])}118        print('Boxed record: ' + str(record))119        return record120    def _unbox_record(self, record):121        """122        Extracts a record from the DynamoDB encapsulation123        """124        for key in record.keys():125            if 'S' in record[key]:126                record[key] = record[key]['S']127            elif 'N' in record[key]:128                if '.' in record[key]['N']:129                    record[key] = float(record[key]['N'])130                else:131                    record[key] = int(record[key]['N'])132            else:133                record[key] = 'unsupported type ' + record[key].keys()[0]134        return record135    136    def _initializeDb(self):137        """138        Initializes the DB creating the table and some test data139        """140        print('Database not initialized. Initializing DB...')141        self.dynamodb_client.create_table(142            AttributeDefinitions = [143                {144                    'AttributeName': 'server_id',145                    'AttributeType': 'S'146                }147            ],148            TableName=self.table_name,149            KeySchema=[150                {151                    'AttributeName': 'server_id',152                    'KeyType': 'HASH'153                }154            ],            155            ProvisionedThroughput={156                'ReadCapacityUnits': 5,157                'WriteCapacityUnits': 5,158            }159        )160        initialized = False161        while not(initialized):162            time.sleep(0.5)163            table_desc = self.dynamodb_client.describe_table(164                TableName = self.table_name)165            initialized = table_desc['Table']['TableStatus'] == 'ACTIVE'166        # Insert some test data167        self.dynamodb_client.batch_write_item(168            RequestItems={169                self.table_name : [170                    {171                        'PutRequest': {172                            'Item': {173                                'server_id': {174                                    'S': 'VillaconejosSQLServer01'175                                },176                                'memory': {177                                    'N': '8589934592'178                                },179                                'iisPresent': {180                                    'S': 'false'181                                }182                            }183                        }184                    },185                    {186                        'PutRequest': {187                            'Item': {188                                'server_id': {189                                    'S': 'VillaconejosSQLServer02'190                                },191                                'memory': {192                                    'N': '8589934592'193                                },194                                'iisPresent': {195                                    'S': 'false'196                                }197                            }198                        }199                    },200                    {201                        'PutRequest': {202                            'Item': {203                                'server_id': {204                                    'S': 'VillaconejosBOSServer01'205                                },206                                'memory': {207                                    'N': '1073741824'208                                },209                                'iisPresent': {210                                    'S': 'true'211                                }212                            }213                        }214                    }215                ]216            }217        )218        print('Database initialized')219# Test it!...

test_class_ddb.py

Source:test_class_ddb.py

1from contextlib import contextmanager2from aws_dynamodb import AWSDynamoDB3@contextmanager4def ddb_table_setup(dynamodb_client):5    dynamodb_client.create_table(6            TableName="Table",7            KeySchema=[8                {9                    'AttributeName': 'key',10                    'KeyType': 'HASH'11                },12                {13                    'AttributeName': 'sort',14                    'KeyType': 'RANGE'15                },16            ],17            AttributeDefinitions=[18                {19                    'AttributeName': 'key',20                    'AttributeType': 'S'21                },22                {23                    'AttributeName': 'sort',24                    'AttributeType': 'S'25                },26            ],27            ProvisionedThroughput={28                'ReadCapacityUnits': 5,29                'WriteCapacityUnits': 530            }31        )32    yield33    table = dynamodb_client.Table("Table")34    table.delete()35@contextmanager36def ddb_table_with_previous_data(dynamodb_client):37    dynamodb_client.create_table(38            TableName="Table",39            KeySchema=[40                {41                    'AttributeName': 'key',42                    'KeyType': 'HASH'43                },44                {45                    'AttributeName': 'sort',46                    'KeyType': 'RANGE'47                },48            ],49            AttributeDefinitions=[50                {51                    'AttributeName': 'key',52                    'AttributeType': 'S'53                },54                {55                    'AttributeName': 'sort',56                    'AttributeType': 'S'57                },58            ],59            ProvisionedThroughput={60                'ReadCapacityUnits': 5,61                'WriteCapacityUnits': 562            }63        )64    table = dynamodb_client.Table("Table")65    table.put_item(Item={66        "key": "key",67        "sort": "sort",68        "data": {}69    })70    yield71    table.delete_item(72        Key={73            "key": "key",74            "sort": ""75        }76    )77    table.delete()78class TestClassDDB:79    def test_table_exist(self, dynamodb_client):80        with ddb_table_setup(dynamodb_client):81            client = AWSDynamoDB()82            client.set_table_name("Table")83            expected = client.table_exist()84            assert expected is True85    def test_table_not_exist(self, dynamodb_client):86        with ddb_table_setup(dynamodb_client):87            client = AWSDynamoDB()88            client.set_table_name("no existing table")89            expected = client.table_exist()90            assert expected is False91    def test_put_item_success(self, dynamodb_client):92        with ddb_table_setup(dynamodb_client):93            data_input = {94                "key": "key_1",95                "sort": "range_1",96                "some_extra_data": "extra_data"97            }98            client = AWSDynamoDB()99            client.set_table_name("Table")100            expected = client.put_item(data_input)101            assert expected == {102                "status": "success",103                "item": data_input104            }105    def test_put_item_fail(self, dynamodb_client):106        with ddb_table_setup(dynamodb_client):107            data_input = {108                "fail_schema": "no data schema"109            }110            client = AWSDynamoDB()111            client.set_table_name("Table")112            expected = client.put_item(data_input)113            assert expected == {114                'status': 'error',115                'message_error': (116                    'An error occurred (ValidationException)'117                    ' when calling the PutItem operation:'118                    ' One or more parameter values were invalid:'119                    ' Missing the key key in the item'120                )121            }122    def test_get_item_success_item_found(self, dynamodb_client):123        with ddb_table_with_previous_data(dynamodb_client):124            client = AWSDynamoDB()125            client.set_table_name("Table")126            expected = client.get_item({"key": "key", "sort": "sort"})127            assert expected == {128                'item': {129                    'data': {},130                    'key': 'key',131                    'sort': 'sort'132                },133                'status': 'success'134            }135    def test_get_item_success_item_not_found(self, dynamodb_client):136        with ddb_table_with_previous_data(dynamodb_client):137            client = AWSDynamoDB()138            client.set_table_name("Table")139            expected = client.get_item({"key": "key", "sort": ""})140            assert expected == {'item': {}, 'status': 'success'}141    def test_get_item_fail(self, dynamodb_client):142        with ddb_table_with_previous_data(dynamodb_client):143            client = AWSDynamoDB()144            client.set_table_name("Table")145            expected = client.get_item({"bad schema": "no value"})146            assert expected == {147                'status': 'error',148                'message_error': (149                    'An error occurred (ValidationException)'150                    ' when calling the GetItem operation:'151                    ' Validation Exception'152                )...

cache.py

Source:cache.py

1import boto32dynamodb_client = boto3.resource('dynamodb', region_name='us-east-2')3kinesis_client = boto3.client('kinesis', region_name='us-east-2')4KINESIS_STREAM_NAME = None5PAYWALLS = dynamodb_client.Table('PaywallBotPaywall').scan()['Items']6EMOJIS = dynamodb_client.Table('PaywallBotEmoji').scan()['Items']7REPLIES = dynamodb_client.Table('PaywallBotReply').scan()['Items']...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.