def _run_sql_for_alerts()

in opmon/monitoring.py [0:0]

• 30 lines of code
• 4 McCabe index (conditional complexity)

    def _run_sql_for_alerts(self, submission_date) -> None:
        try:
            self._check_runnable(submission_date)
        except Exception as e:
            print(f"Failed to run opmon project: {e}")
            return

        total_alerts = 0

        for _ in self.config.alerts:
            total_alerts += 1

        if total_alerts <= 0:
            print(f"No alerts configured for {self.normalized_slug}")
            return

        table_name = f"{self.normalized_slug}_alerts_v{SCHEMA_VERSIONS['alert']}"
        self.bigquery.execute(
            self._get_sql_for_alerts(submission_date=submission_date),
            destination_table=f"{table_name}${submission_date:%Y%m%d}",
            clustering=["build_id"],
            time_partitioning="submission_date",
            write_disposition=bigquery.job.WriteDisposition.WRITE_TRUNCATE,
            dataset=self.derived_dataset,
        )

        print(f"Create alerts view for {self.slug}")
        self.bigquery.execute(
            self._get_alerts_view_sql(),
            annotations={
                "slug": self.slug,
                "type": "alerts_view",
                "submission_date": submission_date,
            },
        )