def insert()

in xlml/utils/bigquery.py [0:0]

• 28 lines of code
• 8 McCabe index (conditional complexity)

  def insert(self, test_runs: Iterable[TestRun]) -> None:
    """Insert Benchmark test runs into the table.

    Args:
      test_runs: Test runs in a benchmark test job.
    """
    for run in test_runs:
      # job hisotry rows
      job_history_rows = [dataclasses.astuple(run.job_history)]

      # metric hisotry rows
      metric_history_rows = []
      for each in run.metric_history:
        if self.is_valid_metric(each.metric_value):
          metric_history_rows.append(dataclasses.astuple(each))
        else:
          logging.error(f"Discarding metric as {each.metric_value} is invalid.")

      # metadata hisotry rows
      metadata_history_rows = []
      for each in run.metadata_history:
        metadata_history_rows.append(dataclasses.astuple(each))

      for table_id, rows in [
          (self.job_history_table_id, job_history_rows),
          (self.metric_history_table_id, metric_history_rows),
          (self.metadata_history_table_id, metadata_history_rows),
      ]:
        if not rows:
          continue
        logging.info(
            f"Inserting {len(rows)} rows into BigQuery table {table_id}."
        )
        table = self.client.get_table(table_id)
        errors = self.client.insert_rows(table, rows)

        if errors:
          raise RuntimeError(f"Failed to add rows to Bigquery: {errors}.")
        else:
          logging.info("Successfully added rows to Bigquery.")