src/dirbs/importer/abstract_importer.py
def _process_pipeline_jobs(self, *, executor, futures_to_type, state, num_batches_finalized=False, timeout=None):
        """Helper function to DRY out processing pipeline futures."""
        total_batches = state['num_batches']
        total_batches_msg = ' of {total:d}'.format(total=total_batches) if num_batches_finalized else ''

        try:
            for future in futures.as_completed(futures_to_type, timeout=timeout):
                job_type = futures_to_type[future]
                del futures_to_type[future]
                if job_type is UploadPipelineJobType.PREVALIDATE:
                    try:
                        res = future.result()  # will throw exception if this one was thrown
                        state['num_validated_batches'] += 1
                        done_batches = state['num_validated_batches']
                        msg = 'Pre-validated {done:d}{total} split files'.format(done=done_batches,
                                                                                 total=total_batches_msg)
                        # Queue upload job now that data has been pre-validated
                        job = executor.submit(self._upload_file_to_staging_table, res)
                        futures_to_type[job] = UploadPipelineJobType.UPLOAD
                    except exceptions.PrevalidationCheckRawException as err:
                        raise exceptions.PrevalidationCheckException(str(err),
                                                                     statsd=self._statsd,
                                                                     metrics_failures_root=self._metrics_failures_root)
                else:
                    future.result()
                    state['num_uploaded_batches'] += 1
                    done_batches = state['num_uploaded_batches']
                    msg = 'Uploaded {done:d}{total} split files'.format(done=done_batches, total=total_batches_msg)

                log_lvl = logging.DEBUG
                if done_batches % 50 == 0 or (done_batches == total_batches and num_batches_finalized):
                    log_lvl = logging.INFO
                self._logger.log(log_lvl, msg)
        except futures.TimeoutError:
            pass
Helper function to DRY out processing pipeline futures.
Helper function to DRY out processing pipeline futures.
Helper function to DRY out processing pipeline futures .
def _process_pipeline_jobs(self, *, executor, futures_to_type, state, num_batches_finalized=False, timeout=None):
        total_batches = state['num_batches']
        total_batches_msg = ' of {total:d}'.format(total=total_batches) if num_batches_finalized else ''
        try:
            for future in futures.as_completed(futures_to_type, timeout=timeout):
                job_type = futures_to_type[future]
                del futures_to_type[future]
                if job_type is UploadPipelineJobType.PREVALIDATE:
                    try:
                        res = future.result()  
                        state['num_validated_batches'] += 1
                        done_batches = state['num_validated_batches']
                        msg = 'Pre-validated {done:d}{total} split files'.format(done=done_batches,
                                                                                 total=total_batches_msg)
                        job = executor.submit(self._upload_file_to_staging_table, res)
                        futures_to_type[job] = UploadPipelineJobType.UPLOAD
                    except exceptions.PrevalidationCheckRawException as err:
                        raise exceptions.PrevalidationCheckException(str(err),
                                                                     statsd=self._statsd,
                                                                     metrics_failures_root=self._metrics_failures_root)
                else:
                    future.result()
                    state['num_uploaded_batches'] += 1
                    done_batches = state['num_uploaded_batches']
                    msg = 'Uploaded {done:d}{total} split files'.format(done=done_batches, total=total_batches_msg)
                log_lvl = logging.DEBUG
                if done_batches % 50 == 0 or (done_batches == total_batches and num_batches_finalized):
                    log_lvl = logging.INFO
                self._logger.log(log_lvl, msg)
        except futures.TimeoutError:
            pass
def _process_pipeline_jobs ( self , * , executor , futures_to_type , state , num_batches_finalized = False , timeout = None ) : total_batches = state [ 'num_batches' ] total_batches_msg = ' of {total:d}' . format ( total = total_batches ) if num_batches_finalized else '' try : for future in futures . as_completed ( futures_to_type , timeout = timeout ) : job_type = futures_to_type [ future ] del futures_to_type [ future ] if job_type is UploadPipelineJobType . PREVALIDATE : try : res = future . result ( ) state [ 'num_validated_batches' ] += 1 done_batches = state [ 'num_validated_batches' ] msg = 'Pre-validated {done:d}{total} split files' . format ( done = done_batches , total = total_batches_msg ) job = executor . submit ( self . _upload_file_to_staging_table , res ) futures_to_type [ job ] = UploadPipelineJobType . UPLOAD except exceptions . PrevalidationCheckRawException as err : raise exceptions . PrevalidationCheckException ( str ( err ) , statsd = self . _statsd , metrics_failures_root = self . _metrics_failures_root ) else : future . result ( ) state [ 'num_uploaded_batches' ] += 1 done_batches = state [ 'num_uploaded_batches' ] msg = 'Uploaded {done:d}{total} split files' . format ( done = done_batches , total = total_batches_msg ) log_lvl = logging . DEBUG if done_batches % 50 == 0 or ( done_batches == total_batches and num_batches_finalized ) : log_lvl = logging . INFO self . _logger . log ( log_lvl , msg ) except futures . TimeoutError : pass
Helper function to DRY out processing pipeline futures.
Helper function to DRY out processing pipeline futures .
"""Helper function to DRY out processing pipeline futures.""" # will throw exception if this one was thrown # Queue upload job now that data has been pre-validated
self
executor
futures_to_type
state
num_batches_finalized
timeout