def invoke()

in source/idea/idea-administrator/src/ideaadministrator/app/delete_cluster.py [0:0]

• 56 lines of code
• 27 McCabe index (conditional complexity)

    def invoke(self):

        # Finding ec2 instances
        self.find_ec2_instances()

        if Utils.is_not_empty(self.ec2_instances):
            self.print_ec2_instances(self.ec2_instances)
            print(f'{len(self.ec2_instances)} ec2 instances will be terminated.')

        # Finding CloudFormation stacks
        self.find_cloud_formation_stacks()
        self.print_cloud_formation_stacks()

        if not self.force:
            confirm = self.context.prompt(f'Are you sure you want to delete cluster: {self.cluster_name}, region: {self.aws_region} ?')
            if not confirm:
                return

        self.invoke_app_app_module_clean_up()

        if Utils.is_not_empty(self.termination_protected_ec2_instances):
            self.print_ec2_instances(self.termination_protected_ec2_instances)
            print(f'found {len(self.termination_protected_ec2_instances)} EC2 instances with termination protection enabled.')
            if not self.force:
                confirm = self.context.prompt('Are you sure you want to disable termination protection for above instances ?')
                if not confirm:
                    return

        self.detach_vpc_from_lambda_functions()

        self.delete_ec2_instances()
        self.delete_cloud_formation_stacks()

        # Delete identity-provider stack - removing UserPool protection
        self.delete_identity_provider_stacks()

        # delete backups if applicable
        if self.delete_backups or self.delete_all:
            confirm_delete_backups = self.force
            if not self.force:
                confirm_delete_backups = self.context.prompt(f'Are you sure you want to delete all the backup recovery points associated with the cluster: 'f'{self.cluster_name}?')

            if confirm_delete_backups:
                self.delete_backup_vault_recovery_points()

        self.delete_cluster_stack()

        if self.delete_bootstrap or self.delete_all:
            confirm_delete_bootstrap = self.force
            if not self.force:
                confirm_delete_bootstrap = self.context.prompt(f'Are you sure you want to delete the bootstrap stack and S3 Bucket associated with the cluster: 'f'{self.cluster_name}? This action is not reversible.')

            if confirm_delete_bootstrap:
                self.delete_bootstrap_and_s3_bucket()

        #Required here because QUIC support modifies the target groups which cloudformation cannot recognize
        #At this point load balancers and listeners have been deleted
        self.context.info(f'Deleting target groups...')
        self.delete_target_groups()
        
        if self.delete_databases or self.delete_all:
            self.find_dynamodb_tables()
            if Utils.is_not_empty(self.dynamodb_tables):
                self.print_dynamodb_tables()

                confirm_delete_databases = self.force
                if not self.force:
                    confirm_delete_databases = self.context.prompt(f'Are you sure you want to delete all dynamodb tables associated with the cluster: 'f'{self.cluster_name}?')

                if confirm_delete_databases:
                    self.delete_dynamodb_tables()

        if self.delete_cloudwatch_logs or self.delete_all:
            self.find_cloudwatch_logs()
            if Utils.is_not_empty(self.cloudwatch_logs):
                self.print_cloudwatch_logs()

                confirm_delete_cloudwatch_logs = self.force
                if not self.force:
                    confirm_delete_cloudwatch_logs = self.context.prompt(f'Are you sure you want to delete all cloudwatch logs associated with the cluster: 'f'{self.cluster_name}?')

                if confirm_delete_cloudwatch_logs:
                    self.delete_cloudwatch_log_groups()