application_services.py

from datetime import timezone, datetime
import time

from . import DbTerminator, Terminator


class Cloudformation(Terminator):
    @staticmethod
    def create(credentials):
        def paginate_stacks(client):
            return client.get_paginator('describe_stacks').paginate().build_full_result()['Stacks']

        return Terminator._create(credentials, Cloudformation, 'cloudformation', paginate_stacks)

    @property
    def created_time(self):
        return self.instance['CreationTime']

    @property
    def name(self):
        return self.instance['StackName']

    def terminate(self):
        self.client.update_termination_protection(StackName=self.name, EnableTerminationProtection=False)
        self.client.delete_stack(StackName=self.name)


class CloudWatchLogGroup(Terminator):
    @staticmethod
    def create(credentials):
        return Terminator._create(credentials, CloudWatchLogGroup, 'logs', lambda client: client.describe_log_groups()['logGroups'])

    @property
    def name(self):
        return self.instance['logGroupName']

    @property
    def created_time(self):
        # self.instance['creationTime'] is the number of milliseconds after Jan 1, 1970 00:00:00 UTC
        milliseconds = self.instance['creationTime']
        return datetime.fromtimestamp(milliseconds/1000.0, tz=timezone.utc)

    def terminate(self):
        self.client.delete_log_group(logGroupName=self.name)


class CodeBuild(Terminator):

    @staticmethod
    def create(credentials):
        def paginate_projects(client):
            project_names = client.get_paginator(
                'list_projects').paginate().build_full_result()['projects']

            if not project_names:
                return []

            projects = client.batch_get_projects(
                names=project_names).get('projects', ())
            return [
                {'name': p['name'], 'created': p['created']} for p in projects]

        return Terminator._create(
            credentials, CodeBuild, 'codebuild',
            paginate_projects)

    @property
    def created_time(self):
        return self.instance['created']

    @property
    def id(self):
        return self.instance['name']

    @property
    def name(self):
        return self.instance['name']

    def terminate(self):
        self.client.delete_project(name=self.instance['name'])


class CodeCommitRepository(DbTerminator):
    @staticmethod
    def create(credentials):
        def paginate_repositories(client):
            return client.get_paginator('list_repositories').paginate().build_full_result()['repositories']

        return Terminator._create(credentials, CodeCommitRepository, 'codecommit', paginate_repositories)

    @property
    def id(self):
        return self.instance['repositoryId']

    @property
    def name(self):
        return self.instance['repositoryName']

    def terminate(self):
        self.client.delete_repository(repositoryName=self.name)


class CodePipeline(Terminator):

    @staticmethod
    def create(credentials):
        return Terminator._create(
            credentials, CodePipeline, 'codepipeline',
            lambda client: client.list_pipelines().get('pipelines', ()))

    @property
    def created_time(self):
        return self.instance['created']

    @property
    def id(self):
        return self.instance['name']

    @property
    def name(self):
        return self.instance['name']

    def terminate(self):
        self.client.delete_pipeline(name=self.name)


class Efs(Terminator):
    @staticmethod
    def create(credentials):
        return Terminator._create(credentials, Efs, 'efs', lambda client: client.describe_file_systems()['FileSystems'])

    @property
    def id(self):
        return self.instance['FileSystemId']

    @property
    def name(self):
        return self.instance['Name']

    @property
    def created_time(self):
        return self.instance['CreationTime']

    def terminate(self):
        # Cannot delete file system if in use: delete mounts targets first
        for mount_target in self.client.describe_mount_targets(FileSystemId=self.id)['MountTargets']:
            self.client.delete_mount_target(MountTargetId=mount_target['MountTargetId'])
        self.client.delete_file_system(FileSystemId=self.id)


class KinesisStream(Terminator):
    @staticmethod
    def create(credentials):
        def paginate_streams(client):
            names = client.get_paginator('list_streams').paginate(
                PaginationConfig={
                    'PageSize': 100,
                }
            ).build_full_result()['StreamNames']

            if not names:
                return []

            return [
                client.describe_stream(StreamName=n)['StreamDescription'] for n in names
            ]

        return Terminator._create(credentials, KinesisStream, 'kinesis', paginate_streams)

    @property
    def created_time(self):
        return self.instance['StreamCreationTimestamp']

    @property
    def id(self):
        return self.instance['StreamName']

    @property
    def name(self):
        return self.instance['StreamName']

    @property
    def ignore(self):
        return self.instance['StreamStatus'] == 'DELETING'

    def terminate(self):
        self.client.delete_stream(
            StreamName=self.instance['StreamName'],
            EnforceConsumerDeletion=True
        )


class SesIdentity(DbTerminator):
    @staticmethod
    def create(credentials):
        return Terminator._create(credentials, SesIdentity, 'ses',
                                  lambda client: client.list_identities()['Identities'])

    @property
    def id(self):
        return self.instance

    @property
    def name(self):
        return self.instance

    def terminate(self):
        self.client.delete_identity(Identity=self.id)


class SesReceiptRuleSet(Terminator):
    @staticmethod
    def create(credentials):
        def _paginate_receipt_rule_sets(client):
            results = client.list_receipt_rule_sets()
            next_token = results.pop('NextToken', None)
            while next_token:
                # This operation can be made at most once/second, see
                # https://boto3.readthedocs.io/en/latest/reference/services/ses.html#SES.Client.list_receipt_rule_sets
                time.sleep(1)
                next_rule_sets = client.list_receipt_rule_sets(NextToken=next_token)
                results['RuleSets'].append(next_rule_sets['RuleSets'])
                next_token = next_rule_sets.pop('NextToken', None)
            return results['RuleSets']
        return Terminator._create(credentials, SesReceiptRuleSet, 'ses', _paginate_receipt_rule_sets)

    @property
    def name(self):
        return self.instance['Name']

    @property
    def created_time(self):
        return self.instance['CreatedTimestamp']

    def terminate(self):
        self.client.delete_receipt_rule_set(RuleSetName=self.name)


class Sns(DbTerminator):
    @staticmethod
    def create(credentials):
        return Terminator._create(credentials, Sns, 'sns', lambda client: client.list_topics()['Topics'])

    @property
    def id(self):
        return self.instance['TopicArn']

    @property
    def name(self):
        return self.instance['TopicArn']

    def terminate(self):
        self.client.delete_topic(TopicArn=self.id)


class SqsQueue(DbTerminator):
    @staticmethod
    def create(credentials):
        return Terminator._create(credentials, SqsQueue, 'sqs', lambda client: client.list_queues().get('QueueUrls', []))

    @property
    def id(self):
        return self.instance

    @property
    def name(self):
        return self.instance

    def terminate(self):
        self.client.delete_queue(QueueUrl=self.id)


class SsmParameter(DbTerminator):
    @staticmethod
    def create(credentials):
        return Terminator._create(credentials, SsmParameter, 'ssm', lambda client: client.describe_parameters()['Parameters'])

    @property
    def id(self):
        return self.instance['Name']

    @property
    def name(self):
        return self.instance['Name']

    def terminate(self):
        self.client.delete_parameter(Name=self.id)


class DynamoDb(DbTerminator):

    @staticmethod
    def create(credentials):

        def get_tables(client):
            table_names = client.get_paginator(
                'list_tables').paginate().build_full_result().get('TableNames', ())
            return table_names

        return Terminator._create(credentials, DynamoDb, 'dynamodb', get_tables)

    @property
    def id(self):
        return self.instance

    @property
    def name(self):
        return self.instance

    def terminate(self):
        return self.client.delete_table(TableName=self.instance)


class StepFunctions(Terminator):
    @staticmethod
    def create(credentials):

        def get_state_machines(client):
            state_machines = client.get_paginator(
                'list_state_machines').paginate().build_full_result().get('stateMachines', [])
            return state_machines

        return Terminator._create(credentials, StepFunctions, 'stepfunctions', get_state_machines)

    @property
    def created_time(self):
        return self.instance['creationDate']

    @property
    def name(self):
        return self.instance['stateMachineArn']

    def terminate(self):
        return self.client.delete_state_machine(stateMachineArn=self.name)


class CloudWatchAlarm(DbTerminator):
    @staticmethod
    def create(credentials):
        return Terminator._create(credentials, CloudWatchAlarm, 'cloudwatch', lambda client: client.describe_alarms()['MetricAlarms'])

    @property
    def name(self):
        return self.instance['AlarmName']

    def terminate(self):
        self.client.delete_alarms(AlarmNames=[self.name])


class SsmDocument(Terminator):
    @staticmethod
    def create(credentials):
        def get_ssm_documents(client):
            ssm_documents = client.get_paginator(
                'list_documents').paginate(Filters=[{'Key': 'Owner', 'Values': ['self']}]).build_full_result().get('DocumentIdentifiers', [])
            return ssm_documents

        return Terminator._create(credentials, SsmDocument, 'ssm', get_ssm_documents)

    @property
    def created_time(self):
        return self.instance['CreatedDate']

    @property
    def name(self):
        return self.instance['Name']

    def terminate(self):
        self.client.delete_document(Name=self.name)


class SsmSession(Terminator):
    @staticmethod
    def create(credentials):
        def get_ssm_sessions(client):
            ssm_sessions = client.get_paginator(
                'describe_sessions').paginate(State='Active').build_full_result().get('Sessions', [])
            return ssm_sessions

        return Terminator._create(credentials, SsmSession, 'ssm', get_ssm_sessions)

    @property
    def created_time(self):
        return self.instance['StartDate']

    @property
    def name(self):
        return self.instance['SessionId']

    @property
    def id(self):
        return self.instance['SessionId']

    def terminate(self):
        self.client.terminate_session(SessionId=self.name)


class MqBroker(Terminator):
    @staticmethod
    def create(credentials):
        def get_mq_brokers(client):
            mq_brokers = client.get_paginator(
                'list_brokers').paginate().build_full_result().get('BrokerSummaries', [])
            return mq_brokers

        return Terminator._create(credentials, MqBroker, 'mq', get_mq_brokers)

    @property
    def created_time(self):
        return self.instance['Created']

    @property
    def name(self):
        return self.instance['BrokerName']

    @property
    def id(self):
        return self.instance['BrokerId']

    def terminate(self):
        self.client.delete_broker(BrokerId=self.id)