Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use stack_parameters method in localstack

Best Python code snippet using localstack_python

pre-deploy.py

Source:pre-deploy.py

...22        raise Exception('UserData JSON must include the ParameterStoreIdentifier')23    if 'Parameters' not in user_data:24        raise Exception('UserData JSON must include the Parameters')25    return user_data26def add_stack_parameters(stack, stack_parameters):27    for item in stack['Parameters']:28        key = item['ParameterKey']29        value = item['ParameterValue']30        stack_parameters[key] = value31    return stack_parameters32def get_stack_parameters(stack_name):33    client = boto3.client('cloudformation')34    result = client.describe_stacks(StackName=stack_name)35    # convert them into proper formatting we want36    stack_parameters = {}37    if 'Stacks'  in result:38        stack_parameters = add_stack_parameters(result['Stacks'][0], stack_parameters)39        # add root stack parameters as well for completeness40        if 'RootId' in result['Stacks'][0]:41            root_stack_id = result['Stacks'][0]['RootId']42            root_stack_name = root_stack_id.split('/')[1]43            root_stack_result = client.describe_stacks(StackName=root_stack_name)44            if 'Stacks'  in result:45                stack_parameters = add_stack_parameters(root_stack_result['Stacks'][0], stack_parameters)46    return stack_parameters47def update_parameter(name, description, value, type):48    if value != '':49        ssm.put_parameter(Name=name,Description=description,Value=value,Type=type, Overwrite=True)50    else:51        print(name + ' value is empty')52def delete_parameter(name):53    # even delete fails dont pass the exception54    try:55        ssm.delete_parameter(Name=name)56    except Exception as e:57        str(e)58def update_parameter_store(stack_parameters, user_parameters, identifier):59    for k, v in stack_parameters.items():60        name = identifier + '/' + k61        update_parameter(name, '', v, 'String')62    for k, v in user_parameters.items():63        name = identifier + '/' + k64        update_parameter(name, v['Description'], v['Value'], v['Type'])65def delete_parameter_store(stack_parameters, user_parameters, identifier):66    for k, v in stack_parameters.items():67        name = identifier + '/' + k68        delete_parameter(name)69    for k, v in user_parameters.items():70        name = identifier + '/' + k71        delete_parameter(name)72# setups the parameter store73def setup_parameter_store_handler(user_data, request_type):74    print('handler parameter store started')75    validate_user_data_for_parameter_store(user_data)76    parameter_store_id = user_data['ParameterStoreIdentifier']77    user_parameters = json.loads(user_data['Parameters'])78    # extract cloudformation iam role from user data and set it in user parameters79    if 'BaseStack' in user_data:80        base_stack_name = user_data['BaseStack']81        user_parameters['BaseStack']['Value'] = base_stack_name82    if 'ParentStack' in user_data:83        parent_stack_name = user_data['ParentStack']84        user_parameters['ParentStack']['Value'] = parent_stack_name85    if 'PipelineMasterStackTemplateName' in user_data:86        pipeline_master_stack_template_name = user_data['PipelineMasterStackTemplateName']87        user_parameters['PipelineMasterStackTemplateName']['Value'] = pipeline_master_stack_template_name88    if 'PipelineMasterStackTemplateConfigName' in user_data:89        pipeline_master_stack_template_config_name = user_data['PipelineMasterStackTemplateConfigName']90        user_parameters['PipelineMasterStackTemplateConfigName']['Value'] = pipeline_master_stack_template_config_name91    if 'PipelineCloudformationIAMRoleARN' in user_data:92        pipeline_cfn_iam_role_arn = user_data['PipelineCloudformationIAMRoleARN']93        user_parameters['PipelineCloudformationIAMRoleARN']['Value'] = pipeline_cfn_iam_role_arn94    if 'StackName'  in user_data:95        stack_parameters = get_stack_parameters(user_data['StackName'])96    else:97        stack_parameters = {}98    if request_type == 'Create':99        update_parameter_store(stack_parameters, user_parameters, parameter_store_id)100    elif request_type == 'Update':101        params = exclude_parameters(stack_parameters, user_parameters)102        update_parameter_store(params['stack_params'], params['user_params'], parameter_store_id)103    elif request_type == 'Delete':104        delete_parameter_store(stack_parameters, user_parameters, parameter_store_id)105    print('handler parameter store finished')106# Some parameters should not be updated in to SSM Parameter Store when request_type is update107def exclude_parameters(stack_parameters, user_parameters):108    exclude_params = ['PipelineExecutionVersion', 'EPOAdminUserName', 'EPOAdminPassword', 'ConfirmEPOAdminPassword', 'EPOPassphraseDR' ,'ConfirmEPOPassphraseDR' , 'EPODatabaseName', 'DBMasterUserPassword', 'ConfirmDBMasterUserPassword', 'DBEncrypted', 'DBEncryptionKmsKeyARN']109    user_parameters_copy = copy.deepcopy(user_parameters)...

get_stack_parameters.py

Source:get_stack_parameters.py

1import json2import os3import boto34class StackParametersBuilder:5    def __init__(self, resource_dir):6        self.resource_dir = resource_dir7        self.parameters = {}8        self.redshift_client = None9        self.parameters_fetched_from_stack_details = False10        self.kms_encrypted_password_fetched_from_password_kms_txt = False11        self.are_s3_parameters_enriched = False12        self.are_cluster_parameters_enriched = False13    def get_redshift_client(self):14        if self.redshift_client is None:15            self.get_parameters_from_stack_details()16            self.redshift_client = boto3.client('redshift', region_name=self.parameters['Region'])17        return self.redshift_client18    def get_parameters_from_stack_details(self):19        if not self.parameters_fetched_from_stack_details:20            stack_details_path = '{resource_dir}/STACK_DETAILS.json'.format(resource_dir=self.resource_dir)21            with open(stack_details_path, 'r') as stack_details_json:22                stack_details = json.load(stack_details_json)23                outputs = stack_details['Stacks'][0]['Outputs']24                for output in outputs:25                    output_key = output['OutputKey']26                    output_value = output['OutputValue']27                    self.parameters[output_key] = output_value28                self.parameters['Region'] = stack_details['Stacks'][0]['StackId'].split(':')[3]29            self.parameters_fetched_from_stack_details = True30    def get_kms_encrypted_password_from_password_kms_txt(self):31        if not self.kms_encrypted_password_fetched_from_password_kms_txt:32            password_kms_txt = '{resource_dir}/PASSWORD_KMS.txt'.format(resource_dir=self.resource_dir)33            with open(password_kms_txt, 'r') as kms_encrypted_password_file:34                self.parameters['KMSEncryptedPassword'] = kms_encrypted_password_file.readline()35            self.kms_encrypted_password_fetched_from_password_kms_txt = True36    def enrich_s3_parameters(self):37        if not self.are_s3_parameters_enriched:38            for parameter_key in list(self.parameters):39                if parameter_key.startswith('S3') and parameter_key.endswith('BucketArn'):40                    bucket_arn = self.parameters[parameter_key]41                    new_key = parameter_key[2:-3]42                    self.parameters[new_key] = bucket_arn.replace('arn:aws:s3:::', '')43            self.are_s3_parameters_enriched = True44    def enrich_cluster_parameters(self):45        if not self.are_cluster_parameters_enriched:46            for cluster in ['SourceCluster', 'TargetCluster']:47                cluster_name = self.parameters[cluster+'Name']48                cluster_describe_response = self.get_redshift_client().describe_clusters(ClusterIdentifier=cluster_name)49                if 'Clusters' not in cluster_describe_response or len(cluster_describe_response['Clusters']) != 1:50                    raise Exception('Could not get details for {type} cluster'.format(type=cluster))51                for parameter in ['DBName', 'MasterUsername', 'Endpoint.Address', 'Endpoint.Port']:52                    parameter_parts = parameter.split('.')53                    parameter_value = cluster_describe_response['Clusters'][0]54                    for parameter_part in parameter_parts:55                        parameter_value = parameter_value.get(parameter_part)56                    parameter_parts.insert(0, cluster)57                    self.parameters[''.join(parameter_parts)] = str(parameter_value)58            self.are_cluster_parameters_enriched = True59    def get_parameters_dict(self):60        self.get_parameters_from_stack_details()61        self.get_kms_encrypted_password_from_password_kms_txt()62        self.enrich_s3_parameters()63        self.enrich_cluster_parameters()64        return self.parameters65if __name__ == '__main__':66    default_dir = os.environ['HOME']67    persist_JSON_parameters_target = '{dir}/stack_parameters.json'.format(dir=default_dir)68    persist_BASH_parameters_target = '{dir}/variables.sh'.format(dir=default_dir)69    if os.path.isfile(persist_JSON_parameters_target):70        with open(persist_JSON_parameters_target, 'r') as stack_parameters:71            parameters = json.load(stack_parameters)72    else:73        pb = StackParametersBuilder(default_dir)74        parameters = pb.get_parameters_dict()75        with open(persist_JSON_parameters_target, 'w') as stack_parameters:76            json.dump(parameters, stack_parameters)77        with open(persist_BASH_parameters_target, 'a') as bash_variables:78            for key, value in parameters.items():79                bash_variables.write('{key}="{value}"\n'.format(key=key, value=value.strip()))...

stack.py

Source:stack.py

1# -*- coding: utf-8 -*-2from .pkg.fingerprint import fingerprint3def upload_cf_template(boto_ses,4                       template_content,5                       bucket_name,6                       prefix,7                       format_is_json=True):8    """9    Upload cloudformation template to s3 bucket and returns template url.10    :type boto_ses:11    :type template_content: str12    :type bucket_name: str13    :type prefix: str14    :type format_is_json: bool15    :rtype: str16    """17    s3_client = boto_ses.client("s3")18    fname = fingerprint.of_text(template_content)19    if prefix.endswith("/"):20        prefix = prefix[:-1]21    if format_is_json:22        ext = "json"23    else:24        ext = "yml"25    s3_key = "{}/{}.{}".format(prefix, fname, ext)26    s3_client.put_object(27        Body=template_content,28        Bucket=bucket_name,29        Key=s3_key,30    )31    template_url = "https://s3.amazonaws.com/{}/{}".format(bucket_name, s3_key)32    return template_url33def deploy_stack(boto_ses,34                 stack_name,35                 template_url,36                 stack_tags,37                 stack_parameters,38                 execution_role_arn=None):39    cf_client = boto_ses.client("cloudformation")40    try:41        res = cf_client.describe_stacks(42            StackName=stack_name43        )44        if len(res["Stacks"]) == 1:45            stack_exists_flag = True46        else:47            stack_exists_flag = False48    except:49        stack_exists_flag = False50    if stack_exists_flag is True:51        res = cf_client.update_stack(52            StackName=stack_name,53            TemplateURL=template_url,54            Parameters=stack_parameters,55            Capabilities=[56                "CAPABILITY_NAMED_IAM",57            ],58            # RoleARN=execution_role_arn,59            Tags=stack_tags,60        )61    else:62        res = cf_client.create_stack(63            StackName=stack_name,64            TemplateURL=template_url,65            Parameters=stack_parameters,66            Capabilities=[67                "CAPABILITY_NAMED_IAM",68            ],69            # RoleARN=execution_role_arn,70        )71    return res72def deploy_stack_set(boto_ses,73                     stack_set_name,74                     template_url,75                     stack_tags,76                     stack_parameters,77                     stack_set_admin_role,78                     accounts,79                     regions):80    cf_client = boto_ses.client("cloudformation")81    try:82        res_describe_stack_set = cf_client.describe_stack_set(83            StackSetName=stack_set_name84        )85        if res_describe_stack_set["StackSet"]["StackSetName"] == stack_set_name:86            stack_set_exists_flag = True87        else:88            stack_set_exists_flag = False89    except:90        stack_set_exists_flag = False91    if stack_set_exists_flag is False:92        res_create_stack_set = cf_client.create_stack_set(93            StackSetName=stack_set_name,94            TemplateURL=template_url,95            Parameters=stack_parameters,96            Tags=stack_tags,97            Capabilities=[98                "CAPABILITY_NAMED_IAM",99            ],100            AdministrationRoleARN=stack_set_admin_role,101            ExecutionRoleName="AWSCloudFormationStackSetExecutionRole",102        )103    res_list_stack_instances = cf_client.list_stack_instances(104        StackSetName=stack_set_name105    )106    if len(res_list_stack_instances["Summaries"]) == 0:107        has_stack_set_instance_flag = False108    else:109        has_stack_set_instance_flag = True110    if has_stack_set_instance_flag:111        res_update_stack_set = cf_client.update_stack_set(112            StackSetName=stack_set_name,113            TemplateURL=template_url,114            Parameters=stack_parameters,115            Capabilities=[116                "CAPABILITY_NAMED_IAM",117            ],118            AdministrationRoleARN=stack_set_admin_role,119            ExecutionRoleName="AWSCloudFormationStackSetExecutionRole",120            Accounts=accounts,121            Regions=regions,122        )123        return res_update_stack_set124    else:125        res_create_stack_instances = cf_client.create_stack_instances(126            StackSetName=stack_set_name,127            Accounts=accounts,128            Regions=regions,129            ParameterOverrides=stack_parameters,130        )...

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.