

def run(self):
    'Run the pipeline.'

    def _run_valid(process_valid, run_valid):
        'Set/maintain the valid state of the run.'
        if ((not process_valid) and run_valid):
            return False
        return run_valid
    valid = True
    for processor in self.pipeline:
        if isinstance(self.data, datatable.DataTable):
            (_valid, _, self.data) = processor.run(self.data, is_table=True, encoding=self.encoding, decode_strategy=self.decode_strategy)
        else:
            (_valid, _, self.data) = processor.run(self.data_source, is_table=False, decode_strategy=processor.decode_strategy, encoding=self.encoding, format=self.format)
        valid = _run_valid(_valid, valid)
        if ((not valid) and self.break_on_invalid_processor):
            break
        if self.data:
            self.data.replay()
    self.set_report_meta()
    if self.post_task:
        self.post_task(self)
    return (valid, self.report)

dataset/ETHPy150Open frictionlessdata/goodtables/goodtables/pipeline/pipeline.py Pipeline.run/VarMisuse@136/182 `self`->`processor`