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

How to use dynamodb method in localstack

Best Python code snippet using localstack_python

utils_dynamodb.py

Source:utils_dynamodb.py

1import json2import logging3from datetime import date, datetime4import sys5import boto36from botocore.exceptions import ClientError7# Default configs8AWS_REGION = 'us-east-1'9AWS_PROFILE = 'localstack'10ENDPOINT_URL = 'http://localhost:4566'11# Nome da tabela12TABLENAME = 'poc-python-dynamodb'13# Specifies the Resource Tags (Optional)14TAGS = [15    {16        'Key': 'Name',17        'Value': 'hands-on-cloud-dynamodb-table'18    }19]20# Specifies the attributes that make up the primary key for a table or an index21KEYSCHEMA = [22    {23        'AttributeName': 'Name',24        'KeyType': 'HASH'25    },26    {27        'AttributeName': 'Email',28        'KeyType': 'RANGE'29    }30]31# Specifies a list of attributes that describe the key schema for the table and indexes32ATTRIBUTEDEFINITIONS = [33    {34        'AttributeName': 'Name',35        'AttributeType': 'S'36    },37    {38        'AttributeName': 'Email',39        'AttributeType': 'S'40    }41]42# Add Table Item and Read Table Item43name = 'hands-on-cloud'44email = 'example@cloud.com'45# Update Table Item46phone_number = '123-456-1234'47boto3.setup_default_session(profile_name=AWS_PROFILE)48# logger config49logger = logging.getLogger()50logging.basicConfig(level=logging.INFO,51                    format='%(asctime)s: %(levelname)s: %(message)s')52def get_boto3_client(service):53    """54    Initialize Boto3 client.55    """56    try:57        client = boto3.client(58            service,59            region_name=AWS_REGION,60            endpoint_url=ENDPOINT_URL61        )62    except Exception as e:63        logger.exception('Error while connecting to LocalStack.')64        raise e65    else:66        return client67def get_boto3_resource(service):68    """69    Initialize Boto3 client.70    """71    try:72        resource = boto3.resource(73            service,74            region_name=AWS_REGION,75            endpoint_url=ENDPOINT_URL76        )77    except Exception as e:78        logger.exception('Error while connecting to LocalStack.')79        raise e80    else:81        return resource82def json_datetime_serializer(obj):83    """84    Helper method to serialize datetime fields85    """86    if isinstance(obj, (datetime, date)):87        return obj.isoformat()88    raise TypeError("Type %s not serializable" % type(obj))89def create_dynamodb_table(table_name):90    """91    Creates a DynamoDB table.92    """93    try:94        dynamodb_client = get_boto3_client("dynamodb")95        response = dynamodb_client.create_table(96            TableName=table_name,97            KeySchema=KEYSCHEMA,98            AttributeDefinitions=ATTRIBUTEDEFINITIONS,99            ProvisionedThroughput={100                'ReadCapacityUnits': 1,101                'WriteCapacityUnits': 1102            },103            Tags=TAGS)104    except ClientError:105        logger.exception('Could not create the table.')106        raise107    else:108        return response109def delete_dynamodb_table(table_name):110    """111    Creates a DynamoDB table.112    """113    try:114        dynamodb_client = get_boto3_client("dynamodb")115        response = dynamodb_client.delete_table(116            TableName=table_name117        )118    except ClientError:119        logger.exception('Could not delete the table.')120        raise121    else:122        return response123def add_dynamodb_table_item(table_name, name, email):124    """125    adds a DynamoDB table.126    """127    try:128        dynamodb_resource = get_boto3_resource("dynamodb")129        table = dynamodb_resource.Table(table_name)130        response = table.put_item(131            Item={132                'Name': name,133                'Email': email134            }135        )136    except ClientError:137        logger.exception('Could not add the item to table.')138        raise139    else:140        return response141def read_dynamodb_table_item(table_name, name, email):142    """143    Reads from a DynamoDB table.144    """145    try:146        dynamodb_resource = get_boto3_resource("dynamodb")147        table = dynamodb_resource.Table(table_name)148        response = table.get_item(149            Key={150                'Name': name,151                'Email': email152            }153        )154    except ClientError:155        logger.exception('Could not read the item from table.')156        raise157    else:158        return response159def update_dynamodb_table_item(table_name, name, email, phone_number):160    """161    update the DynamoDB table item.162    """163    try:164        dynamodb_resource = get_boto3_resource("dynamodb")165        table = dynamodb_resource.Table(table_name)166        response = table.update_item(167            Key={168                'Name': name,169                'Email': email170            },171            UpdateExpression="set phone_number=:ph",172            ExpressionAttributeValues={173                ':ph': phone_number174            }175        )176    except ClientError:177        logger.exception('Could not update the item.')178        raise179    else:180        return response181def delete_dynamodb_table_item(table_name, name, email):182    """183    Deletes the DynamoDB table item.184    """185    try:186        dynamodb_resource = get_boto3_resource("dynamodb")187        table = dynamodb_resource.Table(table_name)188        response = table.delete_item(189            Key={190                'Name': name,191                'Email': email192            }193        )194    except ClientError:195        logger.exception('Could not delete the item.')196        raise197    else:198        return response199def main():200    """201    Main invocation function.202    """203    if sys.argv[1] == 'createtable':204        logger.info('Creating a DynamoDB table...')205        dynamodb = create_dynamodb_table(TABLENAME)206        logger.info(207        f'DynamoDB table created: {json.dumps(dynamodb, indent=4, default=json_datetime_serializer)}')208    elif sys.argv[1] == 'deletetable':209        logger.info('Deleteing DynamoDB table...')210        dynamodb = delete_dynamodb_table(TABLENAME)211        logger.info(212            f'Details: {json.dumps(dynamodb, indent=4, default=json_datetime_serializer)}')213    elif sys.argv[1] == 'addtableitem':214        logger.info('Adding item...')215        dynamodb = add_dynamodb_table_item(TABLENAME, name, email)216        logger.info(217            f'DynamoDB table item created: {json.dumps(dynamodb, indent=4)}')218    elif sys.argv[1] == 'readtableitem':219        logger.info('Reading item...')220        dynamodb = read_dynamodb_table_item(TABLENAME, name, email)221        logger.info(222            f'Item details: {json.dumps(dynamodb, indent=4)}')223    elif sys.argv[1] == 'updatetableitem':224        logger.info('updateing item...')225        dynamodb = update_dynamodb_table_item(226            TABLENAME, name, email, phone_number)227        logger.info(228            f'Item details: {json.dumps(dynamodb, indent=4)}')229    elif sys.argv[1] == 'deletetableitem':230        logger.info('Deleteing item...')231        dynamodb = delete_dynamodb_table_item(232            TABLENAME, name, email)233        logger.info(234            f'Details: {json.dumps(dynamodb, indent=4)}')235    else: 236        print('Invalid command... "'+sys.argv[1]+'"')237if __name__ == '__main__':...

flask_app_stack.py

Source:flask_app_stack.py

...17        self.config = Config(self, 'Config', sys_env=sys_env, _aws_env=kwargs.get('env'))18        self.vpc = self.get_existing_vpc()19        self.cluster = self.get_existing_cluster()20        self.add_namespace_and_service_account()21        self.dynamodb = self.create_dynamodb()22    def add_namespace_and_service_account(self):23        # flask-backendãDynamoDBã«ã¢ã¯ã»ã¹ã§ããServiceAccountãè¿½å 24        # Service Accountã«å¿è¦ãªNamespaceãç»é²25        # flask-frontendã®namespaceã¯ä½æããargocdã«ä½æãã¦ããããã¨ã¨ããã26        # Namespace of flask-backend27        namespace_manifest = {28            'apiVersion': 'v1',29            'kind': 'Namespace',30            'metadata': {31                'name': self.config.flask_app.namespace,32                'labels': {33                    'name': self.config.flask_app.namespace34                }35            }36        }37        namespace = self.cluster.add_manifest('FlaskBackendNamespace',38                                              namespace_manifest)39        # Service Account for flask-backend40        backend_sa = self.cluster.add_service_account(41            'FlaskBackendSA',  # ãã®ååãIAM Roleåã«ä»å ããã42            name=self.config.flask_backend.service_account,  # flask_backend43            namespace=self.config.flask_backend.namespace  # flask-backend44        )45        backend_sa.node.add_dependency(namespace)46        dynamodb_full_access_policy_statements = [47            {48                "Effect": "Allow",49                "Action": [50                    "dynamodb:List*",51                    "dynamodb:DescribeReservedCapacity*",52                    "dynamodb:DescribeLimits",53                    "dynamodb:DescribeTimeToLive"54                ],55                "Resource": ["*"]56            },57            {58                "Effect": "Allow",59                "Action": [60                    "dynamodb:BatchGet*",61                    "dynamodb:DescribeStream",62                    "dynamodb:DescribeTable",63                    "dynamodb:Get*",64                    "dynamodb:Query",65                    "dynamodb:Scan",66                    "dynamodb:BatchWrite*",67                    "dynamodb:CreateTable",68                    "dynamodb:Delete*",69                    "dynamodb:Update*",70                    "dynamodb:PutItem"71                ],72                # "Resource": [self.dynamodb.table_arn]73                "Resource": ["arn:aws:dynamodb:*:*:table/messages"]74            }75        ]76        for statement in dynamodb_full_access_policy_statements:77            backend_sa.add_to_principal_policy(78                aws_iam.PolicyStatement.from_json(statement)79            )80    def get_existing_vpc(self) -> aws_ec2.Vpc:81        return aws_ec2.Vpc.from_lookup(self, 'Vpc', vpc_name=self.config.vpc.name)82    def get_existing_cluster(self) -> aws_eks.Cluster:83        pass84        _env = self.config.env.name  # dev85        _vpc = self.vpc86        # eks cluster attributes87        _cluster_name = aws_cdk.Fn.import_value(f'EksClusterName-{_env}')  # ClusterName-dev88        _kubectl_role_arn = aws_cdk.Fn.import_value(f'EksClusterKubectlRoleArn-{_env}')  # KubectlRoleArn-dev89        _kubectl_security_group_id = aws_cdk.Fn.import_value(f'EksClusterKubectlSecurityGroupId-{_env}')  # KubectlSecurityGroupId-dev90        _oidc_provider_arn = aws_cdk.Fn.import_value(f'EksClusterOidcProviderArn-{_env}')  # OidcProviderArn-dev91        _oidc_provider = aws_eks.OpenIdConnectProvider.from_open_id_connect_provider_arn(92            self,93            'OidcProvider',94            open_id_connect_provider_arn=_oidc_provider_arn)95        cluster = aws_eks.Cluster.from_cluster_attributes(96            self,97            'GetCluster',98            cluster_name=_cluster_name,99            open_id_connect_provider=_oidc_provider,100            kubectl_role_arn=_kubectl_role_arn,101            kubectl_security_group_id=_kubectl_security_group_id,102            vpc=_vpc103        )104        return cluster105    def create_dynamodb(self) -> aws_dynamodb.Table:106        # --------------------------------------------------------------107        #108        # DynamoDB109        #110        # --------------------------------------------------------------111        _dynamodb = aws_dynamodb.Table(112            self,113            id='DynamoDbTable',114            table_name=self.config.flask_backend.dynamodb_table,115            partition_key=aws_dynamodb.Attribute(116                name=self.config.flask_backend.dynamodb_partition,117                type=aws_dynamodb.AttributeType.STRING),118            read_capacity=1,119            write_capacity=1,...

dynamodb.py

Source:dynamodb.py

...6            "aws_secret_access_key": os.environ['AWS_SECRET_ACCESS_KEY'],7            "aws_access_key_id":os.environ['AWS_ACCESS_KEY_ID']8}910def create_in_dynamodb(connection_user, dynamodb_table):11    dynamodb = boto3.resource('dynamodb', **resource)12    id = {"user_id": connection_user.pop("user_id")}1314    if(get_from_dynamodb(id, dynamodb_table, dynamodb)):15        return update_dynamo_item(id, connection_user, dynamodb_table, dynamodb)16    else:17        connection_user.update(id)18        table = dynamodb.Table(dynamodb_table)19        response = table.put_item(20        Item={21                **connection_user22            }23        )24        return response252627def get_from_dynamodb(query_data, dynamodb_table, dynamodb=None):28    if not dynamodb:29        dynamodb = boto3.resource('dynamodb')3031    table = dynamodb.Table(dynamodb_table)3233    try:34        response = table.get_item(Key={**query_data})35    except ClientError as e:36        print(e.response['Error']['Message'])37    else:38        return "Item" in response3940# def update_identifier_list     41
...

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.