src/dirbs/listgen/generator.py
def _perform_sanity_checks(self, base_run_id=None):
        """Method to perform sanity checks on list gen."""
        current_loopback_days = self._lookback_days
        current_blocking_conditions = [c.as_dict() for c in self._config.conditions if c.blocking]
        current_operator_configs = [op.as_dict() for op in self._operators]
        current_amnesty_configs = self._amnesty.as_dict()

        if base_run_id:
            base_successful_job_run = metadata.query_for_command_runs(self._metadata_conn,
                                                                      'dirbs-listgen',
                                                                      successful_only=True,
                                                                      run_id=base_run_id)
            base_successful_job_run = base_successful_job_run[0]
            base_successful_job_run = base_successful_job_run.extra_metadata

        else:
            base_successful_job_run = metadata.query_for_command_runs(self._metadata_conn,
                                                                      'dirbs-listgen',
                                                                      successful_only=True)
            base_successful_job_run = base_successful_job_run[0]
            base_successful_job_run = base_successful_job_run.extra_metadata

        if current_loopback_days == base_successful_job_run['lookback_days'] and \
                current_blocking_conditions == base_successful_job_run['blocking_conditions'] and \
                current_operator_configs == base_successful_job_run['operators'] and \
                current_amnesty_configs == base_successful_job_run['amnesty']:
            return True
        return False
Method to perform sanity checks on list gen.
Method to perform sanity checks on list gen.
Method to perform sanity checks on list gen .
def _perform_sanity_checks(self, base_run_id=None):
        current_loopback_days = self._lookback_days
        current_blocking_conditions = [c.as_dict() for c in self._config.conditions if c.blocking]
        current_operator_configs = [op.as_dict() for op in self._operators]
        current_amnesty_configs = self._amnesty.as_dict()
        if base_run_id:
            base_successful_job_run = metadata.query_for_command_runs(self._metadata_conn,
                                                                      'dirbs-listgen',
                                                                      successful_only=True,
                                                                      run_id=base_run_id)
            base_successful_job_run = base_successful_job_run[0]
            base_successful_job_run = base_successful_job_run.extra_metadata
        else:
            base_successful_job_run = metadata.query_for_command_runs(self._metadata_conn,
                                                                      'dirbs-listgen',
                                                                      successful_only=True)
            base_successful_job_run = base_successful_job_run[0]
            base_successful_job_run = base_successful_job_run.extra_metadata
        if current_loopback_days == base_successful_job_run['lookback_days'] and \
                current_blocking_conditions == base_successful_job_run['blocking_conditions'] and \
                current_operator_configs == base_successful_job_run['operators'] and \
                current_amnesty_configs == base_successful_job_run['amnesty']:
            return True
        return False
def _perform_sanity_checks ( self , base_run_id = None ) : current_loopback_days = self . _lookback_days current_blocking_conditions = [ c . as_dict ( ) for c in self . _config . conditions if c . blocking ] current_operator_configs = [ op . as_dict ( ) for op in self . _operators ] current_amnesty_configs = self . _amnesty . as_dict ( ) if base_run_id : base_successful_job_run = metadata . query_for_command_runs ( self . _metadata_conn , 'dirbs-listgen' , successful_only = True , run_id = base_run_id ) base_successful_job_run = base_successful_job_run [ 0 ] base_successful_job_run = base_successful_job_run . extra_metadata else : base_successful_job_run = metadata . query_for_command_runs ( self . _metadata_conn , 'dirbs-listgen' , successful_only = True ) base_successful_job_run = base_successful_job_run [ 0 ] base_successful_job_run = base_successful_job_run . extra_metadata if current_loopback_days == base_successful_job_run [ 'lookback_days' ] and current_blocking_conditions == base_successful_job_run [ 'blocking_conditions' ] and current_operator_configs == base_successful_job_run [ 'operators' ] and current_amnesty_configs == base_successful_job_run [ 'amnesty' ] : return True return False
Method to perform sanity checks on list gen.
Method to perform sanity checks on list gen .
"""Method to perform sanity checks on list gen."""
self
base_run_id