src/dirbs/importer/base_delta_importer.py
def _check_for_conflicts(self):
        """Check whether this list contains any conflicts (same PK, different extra field values)."""
        metric_key = 'conflicts'
        if self._extra_field_names:
            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._check_for_conflicts_single_partition, name))
                else:
                    futures_list.append(executor.submit(self._check_for_conflicts_single_partition,
                                                        self._staging_tbl_name))

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

            if not conflict_rows:
                return True, 'Conflicting rows check passed', metric_key

            confl_rows_sum = 0
            for x in conflict_rows:
                self._logger.debug('Found {count} '
                                   'conflicting row(s) with primary key {pk_names}: {pk_values}'
                                   .format(count=x.dc,
                                           pk_names=tuple(self._pk_field_names),
                                           pk_values=tuple(getattr(x, pk) for pk in self._pk_field_names)))
                confl_rows_sum += x.dc
            return False, 'Conflicting rows check failed ({0:d} rows with same primary key and conflicting data)' \
                .format(confl_rows_sum), metric_key

        return True, 'Conflicting rows check skipped due to lack of extra_fields', metric_key
Check whether this list contains any conflicts (same PK, different extra field values).
Check whether this list contains any conflicts (same PK, different extra field values).
Check whether this list contains any conflicts ( same PK different extra field values ) .
def _check_for_conflicts(self):
        metric_key = 'conflicts'
        if self._extra_field_names:
            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._check_for_conflicts_single_partition, name))
                else:
                    futures_list.append(executor.submit(self._check_for_conflicts_single_partition,
                                                        self._staging_tbl_name))
            conflict_rows = []
            for f in futures_list:
                partial_conflicts = f.result()
                conflict_rows.extend(partial_conflicts)
            if not conflict_rows:
                return True, 'Conflicting rows check passed', metric_key
            confl_rows_sum = 0
            for x in conflict_rows:
                self._logger.debug('Found {count} '
                                   'conflicting row(s) with primary key {pk_names}: {pk_values}'
                                   .format(count=x.dc,
                                           pk_names=tuple(self._pk_field_names),
                                           pk_values=tuple(getattr(x, pk) for pk in self._pk_field_names)))
                confl_rows_sum += x.dc
            return False, 'Conflicting rows check failed ({0:d} rows with same primary key and conflicting data)' \
                .format(confl_rows_sum), metric_key
        return True, 'Conflicting rows check skipped due to lack of extra_fields', metric_key
def _check_for_conflicts ( self ) : metric_key = 'conflicts' if self . _extra_field_names : 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 . _check_for_conflicts_single_partition , name ) ) else : futures_list . append ( executor . submit ( self . _check_for_conflicts_single_partition , self . _staging_tbl_name ) ) conflict_rows = [ ] for f in futures_list : partial_conflicts = f . result ( ) conflict_rows . extend ( partial_conflicts ) if not conflict_rows : return True , 'Conflicting rows check passed' , metric_key confl_rows_sum = 0 for x in conflict_rows : self . _logger . debug ( 'Found {count} ' 'conflicting row(s) with primary key {pk_names}: {pk_values}' . format ( count = x . dc , pk_names = tuple ( self . _pk_field_names ) , pk_values = tuple ( getattr ( x , pk ) for pk in self . _pk_field_names ) ) ) confl_rows_sum += x . dc return False , 'Conflicting rows check failed ({0:d} rows with same primary key and conflicting data)' . format ( confl_rows_sum ) , metric_key return True , 'Conflicting rows check skipped due to lack of extra_fields' , metric_key
Check whether this list contains any conflicts (same PK, different extra field values).
Check whether this list contains any conflicts ( same PK different extra field values ) .
"""Check whether this list contains any conflicts (same PK, different extra field values).""" # All futures should be done at this point as with block is exited above
self