src/dirbs/importer/abstract_importer.py
def import_data(self):
        """Main import function that checks files and validates the contents before importing to the database."""
        assert self._was_entered and 'Attempting to import with an importer outside of a with statement!'
        try:
            # Store time so that we can track metrics for total listgen time
            st = time.time()

            # Store initial metadata
            metadata.add_optional_job_metadata(self._metadata_conn,
                                               'dirbs-import',
                                               self.import_id,
                                               **self._import_metadata)

            # Log initial message
            self._logger.info("Importing {0} data from file \'{1}\'"
                              .format(self._import_type, self._filename))

            # Init staging table (commit afterwards to ensure other processes can see table)
            with self._conn:
                self._time_component_perf('init_staging', self._init_staging_table)
                if self._supports_imei_shards:
                    self._time_component_perf('init_staging_shards', self._init_staging_table_shards)

            # Compute MD5 hash
            self._time_component_perf('compute_md5', self._compute_md5_hash)
            # Now do extract -> split -> preprocess -> prevalidate -> upload pipeline
            self._time_component_perf('upload_pipeline', self._upload_pipeline)
            # ANALYZE staging table after upload
            self._time_component_perf('analyze_staging', self._analyze_staging_table)
            # Run binary (yes/no) validation checks that operator on "raw" data (prior to post-processing)
            self._time_component_perf('validation_binary_checks_raw', self._validate_binary_checks_raw)
            # Post-process staging table
            self._time_component_perf('postprocess_staging', self._postprocess_staging_data)
            # Run binary (yes/no) validation checks
            self._time_component_perf('validation_binary_checks', self._validate_binary_checks)
            # Run row threshold validation checks
            self._time_component_perf('validation_threshold_checks', self._validate_threshold_checks)
            # Run validation checks based on historic data
            self._time_component_perf('validation_historical_checks', self._validate_historical_checks)
            # Copy data from the staging table
            rows_before = -1  # Sentinel value
            if self._need_previous_count_for_stats:
                rows_before = self.row_count
            rows_inserted, rows_updated, row_deleted = \
                self._time_component_perf('copy_from_staging', self._copy_staging_data)
            # Output import stats
            self._time_component_perf('output_stats', self._output_stats, rows_before, rows_inserted, rows_updated,
                                      row_deleted)

        finally:
            dt = int((time.time() - st) * 1000)
            self._log_normalized_import_time_metrics(dt)
Main import function that checks files and validates the contents before importing to the database.
Main import function that checks files and validates the contents before importing to the database.
Main import function that checks files and validates the contents before importing to the database .
def import_data(self):
        assert self._was_entered and 'Attempting to import with an importer outside of a with statement!'
        try:
            st = time.time()
            metadata.add_optional_job_metadata(self._metadata_conn,
                                               'dirbs-import',
                                               self.import_id,
                                               **self._import_metadata)
            self._logger.info("Importing {0} data from file \'{1}\'"
                              .format(self._import_type, self._filename))
            with self._conn:
                self._time_component_perf('init_staging', self._init_staging_table)
                if self._supports_imei_shards:
                    self._time_component_perf('init_staging_shards', self._init_staging_table_shards)
            self._time_component_perf('compute_md5', self._compute_md5_hash)
            self._time_component_perf('upload_pipeline', self._upload_pipeline)
            self._time_component_perf('analyze_staging', self._analyze_staging_table)
            self._time_component_perf('validation_binary_checks_raw', self._validate_binary_checks_raw)
            self._time_component_perf('postprocess_staging', self._postprocess_staging_data)
            self._time_component_perf('validation_binary_checks', self._validate_binary_checks)
            self._time_component_perf('validation_threshold_checks', self._validate_threshold_checks)
            self._time_component_perf('validation_historical_checks', self._validate_historical_checks)
            rows_before = -1  
            if self._need_previous_count_for_stats:
                rows_before = self.row_count
            rows_inserted, rows_updated, row_deleted = \
                self._time_component_perf('copy_from_staging', self._copy_staging_data)
            self._time_component_perf('output_stats', self._output_stats, rows_before, rows_inserted, rows_updated,
                                      row_deleted)
        finally:
            dt = int((time.time() - st) * 1000)
            self._log_normalized_import_time_metrics(dt)
def import_data ( self ) : assert self . _was_entered and 'Attempting to import with an importer outside of a with statement!' try : st = time . time ( ) metadata . add_optional_job_metadata ( self . _metadata_conn , 'dirbs-import' , self . import_id , ** self . _import_metadata ) self . _logger . info ( "Importing {0} data from file \'{1}\'" . format ( self . _import_type , self . _filename ) ) with self . _conn : self . _time_component_perf ( 'init_staging' , self . _init_staging_table ) if self . _supports_imei_shards : self . _time_component_perf ( 'init_staging_shards' , self . _init_staging_table_shards ) self . _time_component_perf ( 'compute_md5' , self . _compute_md5_hash ) self . _time_component_perf ( 'upload_pipeline' , self . _upload_pipeline ) self . _time_component_perf ( 'analyze_staging' , self . _analyze_staging_table ) self . _time_component_perf ( 'validation_binary_checks_raw' , self . _validate_binary_checks_raw ) self . _time_component_perf ( 'postprocess_staging' , self . _postprocess_staging_data ) self . _time_component_perf ( 'validation_binary_checks' , self . _validate_binary_checks ) self . _time_component_perf ( 'validation_threshold_checks' , self . _validate_threshold_checks ) self . _time_component_perf ( 'validation_historical_checks' , self . _validate_historical_checks ) rows_before = - 1 if self . _need_previous_count_for_stats : rows_before = self . row_count rows_inserted , rows_updated , row_deleted = self . _time_component_perf ( 'copy_from_staging' , self . _copy_staging_data ) self . _time_component_perf ( 'output_stats' , self . _output_stats , rows_before , rows_inserted , rows_updated , row_deleted ) finally : dt = int ( ( time . time ( ) - st ) * 1000 ) self . _log_normalized_import_time_metrics ( dt )
Main import function that checks files and validates the contents before importing to the database.
Main import function that checks files and validates the contents before importing to the database .
"""Main import function that checks files and validates the contents before importing to the database.""" # Store time so that we can track metrics for total listgen time # Store initial metadata # Log initial message # Init staging table (commit afterwards to ensure other processes can see table) # Compute MD5 hash # Now do extract -> split -> preprocess -> prevalidate -> upload pipeline # ANALYZE staging table after upload # Run binary (yes/no) validation checks that operator on "raw" data (prior to post-processing) # Post-process staging table # Run binary (yes/no) validation checks # Run row threshold validation checks # Run validation checks based on historic data # Copy data from the staging table # Sentinel value # Output import stats
self