KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
