def create_changeset()

in awscli/customizations/cloudformation/deployer.py [0:0]
46 lines of code
13 McCabe index (conditional complexity)

    def create_changeset(self, stack_name, cfn_template,
                         parameter_values, capabilities, role_arn,
                         notification_arns, s3_uploader, tags):
        """
        Call Cloudformation to create a changeset and wait for it to complete

        :param stack_name: Name or ID of stack
        :param cfn_template: CloudFormation template string
        :param parameter_values: Template parameters object
        :param capabilities: Array of capabilities passed to CloudFormation
        :param tags: Array of tags passed to CloudFormation
        :return:
        """

        now = datetime.utcnow().isoformat()
        description = "Created by AWS CLI at {0} UTC".format(now)

        # Each changeset will get a unique name based on time
        changeset_name = self.changeset_prefix + str(int(time.time()))

        if not self.has_stack(stack_name):
            changeset_type = "CREATE"
            # When creating a new stack, UsePreviousValue=True is invalid.
            # For such parameters, users should either override with new value,
            # or set a Default value in template to successfully create a stack.
            parameter_values = [x for x in parameter_values
                                if not x.get("UsePreviousValue", False)]
        else:
            changeset_type = "UPDATE"
            # UsePreviousValue not valid if parameter is new
            summary = self._client.get_template_summary(StackName=stack_name)
            existing_parameters = [parameter['ParameterKey'] for parameter in \
                                   summary['Parameters']]
            parameter_values = [x for x in parameter_values
                                if not (x.get("UsePreviousValue", False) and \
                                x["ParameterKey"] not in existing_parameters)]

        kwargs = {
            'ChangeSetName': changeset_name,
            'StackName': stack_name,
            'TemplateBody': cfn_template,
            'ChangeSetType': changeset_type,
            'Parameters': parameter_values,
            'Capabilities': capabilities,
            'Description': description,
            'Tags': tags,
        }

        # If an S3 uploader is available, use TemplateURL to deploy rather than
        # TemplateBody. This is required for large templates.
        if s3_uploader:
            with mktempfile() as temporary_file:
                temporary_file.write(kwargs.pop('TemplateBody'))
                temporary_file.flush()
                url = s3_uploader.upload_with_dedup(
                        temporary_file.name, "template")
                # TemplateUrl property requires S3 URL to be in path-style format
                parts = parse_s3_url(url, version_property="Version")
                kwargs['TemplateURL'] = s3_uploader.to_path_style_s3_url(parts["Key"], parts.get("Version", None))

        # don't set these arguments if not specified to use existing values
        if role_arn is not None:
            kwargs['RoleARN'] = role_arn
        if notification_arns is not None:
            kwargs['NotificationARNs'] = notification_arns
        try:
            resp = self._client.create_change_set(**kwargs)
            return ChangeSetResult(resp["Id"], changeset_type)
        except Exception as ex:
            LOG.debug("Unable to create changeset", exc_info=ex)
            raise ex