src/dirbs/cli/catalog.py
def _prevalidate_file(config, file_descriptor, file_path, file_type, run_id, schema, files_to_delete, logger):
    """
    Pre-validate the input file using the CSV validator.

    :param config: dirbs config instance
    :param file_descriptor: file descriptor
    :param file_path: file path
    :param file_type: file type
    :param run_id: job run id
    :param schema: csv schema to validate with
    :param files_to_delete: list of files to delete
    :param logger: dirbs logger instance
    :return: bool
    """
    split_file_basename = '{0}_import_{1}_split'.format(file_type, run_id)
    split_files = list(split_file(file_descriptor, config.import_config.batch_size,
                                  dirname(file_path), logger, split_file_basename))
    files_to_delete.extend(split_files)
    num_batches = len(split_files)
    num_validated_batches = 0
    with futures.ThreadPoolExecutor(max_workers=config.multiprocessing_config.max_local_cpus) as prevalidator:
        logger.info('Pre-validating file {0} ({1} pre-validation workers)'
                    .format(basename(file_path), config.multiprocessing_config.max_local_cpus))
        tasks = [prevalidator.submit(prevalidate_file, f, schema) for f in split_files]
        for f in futures.as_completed(tasks):
            f.result()  # will throw exception if this one was thrown in thread
            num_validated_batches += 1
            lvl = logging.DEBUG
            if num_validated_batches % 50 == 0 or num_validated_batches == num_batches:
                lvl = logging.INFO  # Only print every 50 batches at INFO so as not to spam console
            logger.log(lvl, 'Pre-validated {validated_batches} of {num_batches} batches'
                       .format(validated_batches=num_validated_batches, num_batches=num_batches))
        logger.info('Successfully pre-validated file: {0}'.format(basename(file_path)))
        return True

    Pre-validate the input file using the CSV validator.

    :param config: dirbs config instance
    :param file_descriptor: file descriptor
    :param file_path: file path
    :param file_type: file type
    :param run_id: job run id
    :param schema: csv schema to validate with
    :param files_to_delete: list of files to delete
    :param logger: dirbs logger instance
    :return: bool
    
Pre-validate the input file using the CSV validator.
Pre - validate the input file using the CSV validator .
def _prevalidate_file(config, file_descriptor, file_path, file_type, run_id, schema, files_to_delete, logger):
    split_file_basename = '{0}_import_{1}_split'.format(file_type, run_id)
    split_files = list(split_file(file_descriptor, config.import_config.batch_size,
                                  dirname(file_path), logger, split_file_basename))
    files_to_delete.extend(split_files)
    num_batches = len(split_files)
    num_validated_batches = 0
    with futures.ThreadPoolExecutor(max_workers=config.multiprocessing_config.max_local_cpus) as prevalidator:
        logger.info('Pre-validating file {0} ({1} pre-validation workers)'
                    .format(basename(file_path), config.multiprocessing_config.max_local_cpus))
        tasks = [prevalidator.submit(prevalidate_file, f, schema) for f in split_files]
        for f in futures.as_completed(tasks):
            f.result()  
            num_validated_batches += 1
            lvl = logging.DEBUG
            if num_validated_batches % 50 == 0 or num_validated_batches == num_batches:
                lvl = logging.INFO  
            logger.log(lvl, 'Pre-validated {validated_batches} of {num_batches} batches'
                       .format(validated_batches=num_validated_batches, num_batches=num_batches))
        logger.info('Successfully pre-validated file: {0}'.format(basename(file_path)))
        return True
def _prevalidate_file ( config , file_descriptor , file_path , file_type , run_id , schema , files_to_delete , logger ) : split_file_basename = '{0}_import_{1}_split' . format ( file_type , run_id ) split_files = list ( split_file ( file_descriptor , config . import_config . batch_size , dirname ( file_path ) , logger , split_file_basename ) ) files_to_delete . extend ( split_files ) num_batches = len ( split_files ) num_validated_batches = 0 with futures . ThreadPoolExecutor ( max_workers = config . multiprocessing_config . max_local_cpus ) as prevalidator : logger . info ( 'Pre-validating file {0} ({1} pre-validation workers)' . format ( basename ( file_path ) , config . multiprocessing_config . max_local_cpus ) ) tasks = [ prevalidator . submit ( prevalidate_file , f , schema ) for f in split_files ] for f in futures . as_completed ( tasks ) : f . result ( ) num_validated_batches += 1 lvl = logging . DEBUG if num_validated_batches % 50 == 0 or num_validated_batches == num_batches : lvl = logging . INFO logger . log ( lvl , 'Pre-validated {validated_batches} of {num_batches} batches' . format ( validated_batches = num_validated_batches , num_batches = num_batches ) ) logger . info ( 'Successfully pre-validated file: {0}' . format ( basename ( file_path ) ) ) return True
Pre-validate the input file using the CSV validator.
Pre - validate the input file using the CSV validator .
"""
    Pre-validate the input file using the CSV validator.

    :param config: dirbs config instance
    :param file_descriptor: file descriptor
    :param file_path: file path
    :param file_type: file type
    :param run_id: job run id
    :param schema: csv schema to validate with
    :param files_to_delete: list of files to delete
    :param logger: dirbs logger instance
    :return: bool
    """ # will throw exception if this one was thrown in thread # Only print every 50 batches at INFO so as not to spam console
None
config
dirbs config instance
dirbs config instance
file_descriptor
None
file_path
None
file_type
None
run_id
job run id
job run id
schema
csv schema to validate with
csv schema to validate with
files_to_delete
list of files to delete
list of files to delete
logger
dirbs logger instance
dirbs logger instance