def run_query(self, query, query_args, query_errors):
    try:
        self.parse_args()
        self.pipeline = beam.Pipeline(options=self.pipeline_options)
        raw_events = self.generate_events()
        query.load(raw_events, query_args)
        result = self.pipeline.run()
        job_duration = self.pipeline_options.view_as(TestOptions).wait_until_finish_duration
        if (self.pipeline_options.view_as(StandardOptions).runner == 'DataflowRunner'):
            result.wait_until_finish(duration=job_duration)
            result.cancel()
        else:
            result.wait_until_finish()
    except Exception as exc:
        query_errors.append(str(exc))
        raise
def run_query(self, query, query_args, query_errors):
    try:
        self.parse_args()
        self.pipeline = beam.Pipeline(options=self.pipeline_options)
        raw_events = self.generate_events()
        query.load(raw_events, query_args)
        result = self.pipeline.run()
        job_duration = self.pipeline_options.view_as(TestOptions).wait_until_finish_duration
        if (self.pipeline_options.view_as(StandardOptions).runner == 'DataflowRunner'):
            result.wait_until_finish(duration=job_duration)
            result.cancel()
        else:
            result.wait_until_finish()
    except Exception as exc:
        query_errors.append(str(exc))
        raise