src/dirbs/importer/base_delta_importer.py
def _delta_multiple_changes_check(self):
        """Check that same record cannot be added or removed at the same time."""
        futures_list = []
        with futures.ThreadPoolExecutor(max_workers=self._max_db_connections) as executor:
            if self._supports_imei_shards:
                for name, rstart, rend in partition_utils.physical_imei_shards(self._conn,
                                                                               tbl_name=self._staging_tbl_name):
                    futures_list.append(executor.submit(self._delta_multiple_changes_check_single_shard, name))
            else:
                futures_list.append(executor.submit(self._delta_multiple_changes_check_single_shard,
                                                    self._staging_tbl_name))

        # All futures should be done at this point as with block is exited above
        failing_rows = []
        for f in futures_list:
            failing_rows.extend(f.result())

        metric_key = 'multiple_changes_check'
        if len(failing_rows) > 0:
            failing_rows_formatted = ', '.join(['{0}: {1}'
                                               .format(pk, failing_rows[0][pk]) for pk in self._pk_field_names])
            return False, 'Same record cannot be added or removed at the same time in delta list. ' \
                          'Failing rows: {fr}...'.format(fr=failing_rows_formatted), metric_key
        return True, 'Conflicting changes check passed', metric_key
Check that same record cannot be added or removed at the same time.
Check that same record cannot be added or removed at the same time.
Check that same record cannot be added or removed at the same time .
def _delta_multiple_changes_check(self):
        futures_list = []
        with futures.ThreadPoolExecutor(max_workers=self._max_db_connections) as executor:
            if self._supports_imei_shards:
                for name, rstart, rend in partition_utils.physical_imei_shards(self._conn,
                                                                               tbl_name=self._staging_tbl_name):
                    futures_list.append(executor.submit(self._delta_multiple_changes_check_single_shard, name))
            else:
                futures_list.append(executor.submit(self._delta_multiple_changes_check_single_shard,
                                                    self._staging_tbl_name))
        failing_rows = []
        for f in futures_list:
            failing_rows.extend(f.result())
        metric_key = 'multiple_changes_check'
        if len(failing_rows) > 0:
            failing_rows_formatted = ', '.join(['{0}: {1}'
                                               .format(pk, failing_rows[0][pk]) for pk in self._pk_field_names])
            return False, 'Same record cannot be added or removed at the same time in delta list. ' \
                          'Failing rows: {fr}...'.format(fr=failing_rows_formatted), metric_key
        return True, 'Conflicting changes check passed', metric_key
def _delta_multiple_changes_check ( self ) : futures_list = [ ] with futures . ThreadPoolExecutor ( max_workers = self . _max_db_connections ) as executor : if self . _supports_imei_shards : for name , rstart , rend in partition_utils . physical_imei_shards ( self . _conn , tbl_name = self . _staging_tbl_name ) : futures_list . append ( executor . submit ( self . _delta_multiple_changes_check_single_shard , name ) ) else : futures_list . append ( executor . submit ( self . _delta_multiple_changes_check_single_shard , self . _staging_tbl_name ) ) failing_rows = [ ] for f in futures_list : failing_rows . extend ( f . result ( ) ) metric_key = 'multiple_changes_check' if len ( failing_rows ) > 0 : failing_rows_formatted = ', ' . join ( [ '{0}: {1}' . format ( pk , failing_rows [ 0 ] [ pk ] ) for pk in self . _pk_field_names ] ) return False , 'Same record cannot be added or removed at the same time in delta list. ' 'Failing rows: {fr}...' . format ( fr = failing_rows_formatted ) , metric_key return True , 'Conflicting changes check passed' , metric_key
Check that same record cannot be added or removed at the same time.
Check that same record cannot be added or removed at the same time .
"""Check that same record cannot be added or removed at the same time.""" # All futures should be done at this point as with block is exited above
self