Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use dynamodb_table_arn method in localstack

Best Python code snippet using localstack_python

create_dynamodb_table.py

Source:create_dynamodb_table.py

1#!/usr/bin/env python2# Copyright 2016-2020 Workiva Inc.3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8#     http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15#16# create_dynamodb_table.py17#18# Script that creates a Dynamodb for use by state machines.19# system imports20import argparse21import logging22import sys23# library imports24# application imports25from aws_lambda_fsm.aws import get_connection26from aws_lambda_fsm.aws import get_arn_from_arn_string27from aws_lambda_fsm.aws import validate_config28from aws_lambda_fsm.constants import ENVIRONMENT_DATA29from aws_lambda_fsm.constants import RETRY_DATA30from aws_lambda_fsm.constants import CHECKPOINT_DATA31from aws_lambda_fsm.constants import CACHE_DATA32from aws_lambda_fsm.constants import STREAM_DATA33from aws_lambda_fsm.constants import AWS_DYNAMODB34from aws_lambda_fsm.constants import AWS35import settings36# setup the command line args37parser = argparse.ArgumentParser(description='Creates AWS DynamoDB tables.')38parser.add_argument('--dynamodb_table_arn', default='PRIMARY_STREAM_SOURCE')39parser.add_argument('--dynamodb_read_capacity_units', type=int, default=10)40parser.add_argument('--dynamodb_write_capacity_units', type=int, default=10)41parser.add_argument('--log_level', default='INFO')42parser.add_argument('--boto_log_level', default='INFO')43args = parser.parse_args()44logging.basicConfig(45    format='[%(levelname)s] %(asctime)-15s %(message)s',46    level=int(args.log_level) if args.log_level.isdigit() else args.log_level,47    datefmt='%Y-%m-%d %H:%M:%S'48)49logging.getLogger('boto3').setLevel(args.boto_log_level)50logging.getLogger('botocore').setLevel(args.boto_log_level)51validate_config()52# setup connections to AWS53dynamodb_table_arn = getattr(settings, args.dynamodb_table_arn)54logging.info('DynamoDB table ARN: %s', dynamodb_table_arn)55logging.info('DynamoDB endpoint: %s', settings.ENDPOINTS.get(AWS.DYNAMODB))56if get_arn_from_arn_string(dynamodb_table_arn).service != AWS.DYNAMODB:57    logging.fatal("%s is not a DynamoDB ARN", dynamodb_table_arn)58    sys.exit(1)59dynamodb_conn = get_connection(dynamodb_table_arn, disable_chaos=True)60dynamodb_table = get_arn_from_arn_string(dynamodb_table_arn).slash_resource()61logging.info('DynamoDB table: %s', dynamodb_table)62if 'RESULTS' in args.dynamodb_table_arn:63    # create a dynamodb table for examples/tracer64    response = dynamodb_conn.create_table(65        TableName=dynamodb_table,66        AttributeDefinitions=[67            {68                AWS_DYNAMODB.AttributeName: 'correlation_id',69                AWS_DYNAMODB.AttributeType: AWS_DYNAMODB.STRING70            },71        ],72        KeySchema=[73            {74                AWS_DYNAMODB.AttributeName: 'correlation_id',75                AWS_DYNAMODB.KeyType: AWS_DYNAMODB.HASH76            }77        ],78        ProvisionedThroughput={79            AWS_DYNAMODB.ReadCapacityUnits: args.dynamodb_read_capacity_units,80            AWS_DYNAMODB.WriteCapacityUnites: args.dynamodb_write_capacity_units81        }82    )83    logging.info(response)84if 'CHECKPOINT' in args.dynamodb_table_arn:85    # create a dynamodb table for checkpoints86    response = dynamodb_conn.create_table(87        TableName=dynamodb_table,88        AttributeDefinitions=[89            {90                AWS_DYNAMODB.AttributeName: CHECKPOINT_DATA.CORRELATION_ID,91                AWS_DYNAMODB.AttributeType: AWS_DYNAMODB.STRING92            },93        ],94        KeySchema=[95            {96                AWS_DYNAMODB.AttributeName: CHECKPOINT_DATA.CORRELATION_ID,97                AWS_DYNAMODB.KeyType: AWS_DYNAMODB.HASH98            }99        ],100        ProvisionedThroughput={101            AWS_DYNAMODB.ReadCapacityUnits: args.dynamodb_read_capacity_units,102            AWS_DYNAMODB.WriteCapacityUnites: args.dynamodb_write_capacity_units103        }104    )105    logging.info(response)106if 'STREAM' in args.dynamodb_table_arn:107    # create a dynamodb table for streaming events108    response = dynamodb_conn.create_table(109        TableName=dynamodb_table,110        AttributeDefinitions=[111            {112                AWS_DYNAMODB.AttributeName: STREAM_DATA.CORRELATION_ID,113                AWS_DYNAMODB.AttributeType: AWS_DYNAMODB.STRING114            },115        ],116        KeySchema=[117            {118                AWS_DYNAMODB.AttributeName: STREAM_DATA.CORRELATION_ID,119                AWS_DYNAMODB.KeyType: AWS_DYNAMODB.HASH120            }121        ],122        ProvisionedThroughput={123            AWS_DYNAMODB.ReadCapacityUnits: args.dynamodb_read_capacity_units,124            AWS_DYNAMODB.WriteCapacityUnites: args.dynamodb_write_capacity_units125        }126    )127    logging.info(response)128if 'RETRY' in args.dynamodb_table_arn:129    # create a dynamodb table for retries. we use partition as the key hash because130    # we want to make the queries for retry entities consistent, and we can easily131    # iterate over a known list (0,...,15) or partitions.132    response = dynamodb_conn.create_table(133        TableName=dynamodb_table,134        AttributeDefinitions=[135            {136                AWS_DYNAMODB.AttributeName: RETRY_DATA.PARTITION,137                AWS_DYNAMODB.AttributeType: AWS_DYNAMODB.NUMBER138            },139            {140                AWS_DYNAMODB.AttributeName: RETRY_DATA.CORRELATION_ID_STEPS,141                AWS_DYNAMODB.AttributeType: AWS_DYNAMODB.STRING142            },143            {144                AWS_DYNAMODB.AttributeName: RETRY_DATA.RUN_AT,145                AWS_DYNAMODB.AttributeType: AWS_DYNAMODB.NUMBER146            },147        ],148        KeySchema=[149            {150                AWS_DYNAMODB.AttributeName: RETRY_DATA.PARTITION,151                AWS_DYNAMODB.KeyType: AWS_DYNAMODB.HASH152            },153            {154                AWS_DYNAMODB.AttributeName: RETRY_DATA.CORRELATION_ID_STEPS,155                AWS_DYNAMODB.KeyType: AWS_DYNAMODB.RANGE156            },157        ],158        LocalSecondaryIndexes=[159            {160                AWS_DYNAMODB.IndexName: RETRY_DATA.RETRIES,161                AWS_DYNAMODB.KeySchema: [162                    {163                        AWS_DYNAMODB.AttributeName: RETRY_DATA.PARTITION,164                        AWS_DYNAMODB.KeyType: AWS_DYNAMODB.HASH165                    },166                    {167                        AWS_DYNAMODB.AttributeName: RETRY_DATA.RUN_AT,168                        AWS_DYNAMODB.KeyType: AWS_DYNAMODB.RANGE169                    }170                ],171                AWS_DYNAMODB.Projection: {172                    AWS_DYNAMODB.ProjectionType: AWS_DYNAMODB.ALL173                }174            }175        ],176        ProvisionedThroughput={177            AWS_DYNAMODB.ReadCapacityUnits: args.dynamodb_read_capacity_units,178            AWS_DYNAMODB.WriteCapacityUnites: args.dynamodb_write_capacity_units179        }180    )181    logging.info(response)182if 'ENVIRONMENT' in args.dynamodb_table_arn:183    # create a dynamodb table storing task environments in order to get around the184    # 8192 character limit185    response = dynamodb_conn.create_table(186        TableName=dynamodb_table,187        AttributeDefinitions=[188            {189                AWS_DYNAMODB.AttributeName: ENVIRONMENT_DATA.GUID,190                AWS_DYNAMODB.AttributeType: AWS_DYNAMODB.STRING191            }192        ],193        KeySchema=[194            {195                AWS_DYNAMODB.AttributeName: ENVIRONMENT_DATA.GUID,196                AWS_DYNAMODB.KeyType: AWS_DYNAMODB.HASH197            }198        ],199        ProvisionedThroughput={200            AWS_DYNAMODB.ReadCapacityUnits: args.dynamodb_read_capacity_units,201            AWS_DYNAMODB.WriteCapacityUnites: args.dynamodb_write_capacity_units202        }203    )204    logging.info(response)205if 'CACHE' in args.dynamodb_table_arn:206    # create a cache table207    response = dynamodb_conn.create_table(208        TableName=dynamodb_table,209        AttributeDefinitions=[210            {211                AWS_DYNAMODB.AttributeName: CACHE_DATA.KEY,212                AWS_DYNAMODB.AttributeType: AWS_DYNAMODB.STRING213            }214        ],215        KeySchema=[216            {217                AWS_DYNAMODB.AttributeName: CACHE_DATA.KEY,218                AWS_DYNAMODB.KeyType: AWS_DYNAMODB.HASH219            }220        ],221        ProvisionedThroughput={222            AWS_DYNAMODB.ReadCapacityUnits: args.dynamodb_read_capacity_units,223            AWS_DYNAMODB.WriteCapacityUnites: args.dynamodb_write_capacity_units224        }225    )...

lambda_fn.py

Source:lambda_fn.py

1import json2import pulumi3import pulumi_aws as aws4from resources.dynamodb import dynamodb_table5def generate_inline_dynamo_policy(dynamodb_table_arn):6    return json.dumps(7        {8            "Version": "2012-10-17",9            "Statement": [10                {11                    "Effect": "Allow",12                    "Action": "dynamodb:UpdateItem",13                    "Resource": f"{dynamodb_table_arn}",14                }15            ],16        }17    )18iam_for_lambda = aws.iam.Role(19    "iamForLambda",20    assume_role_policy="""{21  "Version": "2012-10-17",22  "Statement": [23    {24      "Action": "sts:AssumeRole",25      "Principal": {26        "Service": "lambda.amazonaws.com"27      },28      "Effect": "Allow",29      "Sid": ""30    }31  ]32}33""",34    inline_policies=[35        aws.iam.RoleInlinePolicyArgs(36            name="updateViewCountItem",37            policy=dynamodb_table.arn.apply(38                lambda arn: generate_inline_dynamo_policy(arn)39            ),40        )41    ],42)43lambda_function = aws.lambda_.Function(44    "api_lambda",45    name="resume_api_lambda",46    code=pulumi.FileArchive("../server/"),47    role=iam_for_lambda.arn,48    handler="lambda_function.lambda_handler",49    runtime="python3.9",50    environment=aws.lambda_.FunctionEnvironmentArgs(51        variables={52            "ENV": "production",53        },54    ),...

main.py

Source:main.py

1from init_aws_resources import *2from delete_aws_resources import *3from data_generator import generate_data4import time5# names of aws resources to be created6LAMBDA_ROLE_NAME = 'lambda_role'7STREAM_NAME = 'weather_stream'8TABLE_NAME = 'weather_table'9LAMBDA_FUNCTION_NAME = "save_to_dynomodb"10TIME_TO_LIVE=5*60 11# initialize AWS resources12def init_aws_resources():13    kinesis_arn = create_kinesis_stream(STREAM_NAME)14    print(kinesis_arn)15    dynamodb_table_arn = create_dynamodb_table(TABLE_NAME)16    print(dynamodb_table_arn)17    lambda_role_arn = create_lambda_role(LAMBDA_ROLE_NAME)18    print(lambda_role_arn)19    create_lambda_function(LAMBDA_FUNCTION_NAME, lambda_role_arn, kinesis_arn)20    print("AWS resources are created")21# Delete AWS resources22def delete_aws_resources():23    delete_kinesis_stream(STREAM_NAME)24    delete_lambda_role(LAMBDA_ROLE_NAME)25    delete_dynamodb_table(TABLE_NAME)26    delete_lambda_function(LAMBDA_FUNCTION_NAME)27    print("AWS resources are deleted")28# Main function29def main():30    init_aws_resources()31    generate_data(stream_name=STREAM_NAME)32    # wait N seconds before deleting aws rescources33    time.sleep(TIME_TO_LIVE)34    delete_aws_resources()35if __name__ == "__main__":...

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.