def main()

in tools/domspurge/purge.py [0:0]

• 61 lines of code
• 18 McCabe index (conditional complexity)

def main(args, before, keep_completed, keep_failed, purge_all, recreate):
    log.info('Connecting to Cassandra cluster')

    dc_policy = RoundRobinPolicy()
    token_policy = TokenAwarePolicy(dc_policy)

    if args.username and args.password:
        auth_provider = PlainTextAuthProvider(username=args.username, password=args.password)
    else:
        auth_provider = None

    contact_points = []

    for host_list in args.hosts:
        contact_points.extend(host_list.split(','))

    try:
        with Cluster(contact_points,
                     port=int(args.port),
                     execution_profiles={
                         EXEC_PROFILE_DEFAULT: ExecutionProfile(
                             load_balancing_policy=token_policy,
                             request_timeout=60.0,
                         )
                     },
                     protocol_version=int(args.pv),
                     auth_provider=auth_provider) as cluster:
            session = cluster.connect(args.keyspace) if not recreate else cluster.connect()

            log.info('Connected successfully to Cassandra')

            if recreate:
                log.info('Recreating doms keyspace')
                create_keyspace(session, args.keyspace)
                exit(0)

            log.info('Determining the number of executions that will be dropped')

            execution_count, ids = count_executions(session, before, keep_completed, keep_failed, purge_all)

            if dry_run:
                if execution_count == 0:
                    log.info('No executions will be deleted with the provided criteria')
                elif purge_all:
                    log.info(f'The \'{args.keyspace}\' keyspace will be dropped then recreated w/ all needed tables')
                else:
                    log.info(f'The following executions would be deleted: \n'
                             f'{json.dumps([str(rid) for rid in ids], indent=4)} \n'
                             f'Total: {len(ids):,}')

                exit(0)

            if execution_count == 0 and not purge_all:
                log.info('No executions will be deleted with the provided criteria')
                exit(0)
            elif execution_count == 0 and purge_all:
                if not get_confirmation('No executions will be deleted with the provided criteria. Do you still wish '
                                        f'to drop & recreate the \'{args.keyspace}\' keyspace? [y]/n: '):
                    exit(0)
            else:
                if not get_confirmation(f'{execution_count:,} executions selected for deletion. Continue? [y]/n: '):
                    exit(0)

            if purge_all:
                purge_all_data(session, args.keyspace)
            else:
                for row_id in tqdm(ids, ascii=True, desc='Executions deleted', ncols=80, unit='Execution'):
                    delete_execution(session, row_id)

                log.info(f'Successfully deleted the following executions: \n'
                         f'{json.dumps([str(rid) for rid in ids], indent=4)} \n'
                         f'Total: {len(ids):,}')
    except NoHostAvailable as ne:
        log.exception(ne)
        exit(1)