def run()

in samcli/commands/deploy/deploy_context.py [0:0]

• 53 lines of code
• 11 McCabe index (conditional complexity)

    def run(self):
        """
        Execute deployment based on the argument provided by customers and samconfig.toml.
        """

        # Parse parameters
        with open(self.template_file, "r") as handle:
            template_str = handle.read()

        template_dict = yaml_parse(template_str)

        if not isinstance(template_dict, dict):
            raise deploy_exceptions.DeployFailedError(
                stack_name=self.stack_name, msg="{} not in required format".format(self.template_file)
            )

        parameters = self.merge_parameters(template_dict, self.parameter_overrides)

        template_size = os.path.getsize(self.template_file)
        if template_size > self._max_template_size and not self.s3_bucket:
            raise deploy_exceptions.DeployBucketRequiredError()
        boto_config = get_boto_config_with_user_agent()
        cloudformation_client = boto3.client(
            "cloudformation", region_name=self.region if self.region else None, config=boto_config
        )

        s3_client = None
        if self.s3_bucket:
            s3_client = boto3.client("s3", region_name=self.region if self.region else None, config=boto_config)

            self.s3_uploader = S3Uploader(
                s3_client, self.s3_bucket, self.s3_prefix, self.kms_key_id, self.force_upload, self.no_progressbar
            )

        self.deployer = Deployer(cloudformation_client, client_sleep=self.poll_delay)

        region = s3_client._client_config.region_name if s3_client else self.region  # pylint: disable=W0212
        display_parameter_overrides = hide_noecho_parameter_overrides(template_dict, self.parameter_overrides)
        print_deploy_args(
            self.stack_name,
            self.s3_bucket,
            self.image_repositories if isinstance(self.image_repositories, dict) else self.image_repository,
            region,
            self.capabilities,
            display_parameter_overrides,
            self.confirm_changeset,
            self.signing_profiles,
            self.use_changeset,
            self.disable_rollback,
        )
        return self.deploy(
            self.stack_name,
            template_str,
            parameters,
            self.capabilities,
            self.no_execute_changeset,
            self.role_arn,
            self.notification_arns,
            self.s3_uploader,
            [{"Key": key, "Value": value} for key, value in self.tags.items()] if self.tags else [],
            region,
            self.fail_on_empty_changeset,
            self.confirm_changeset,
            self.use_changeset,
            self.disable_rollback,
        )