def write()

in opmon/metadata.py [0:0]
56 lines of code
12 McCabe index (conditional complexity)

    def write(self) -> None:
        """Update the BQ table with project metadata."""
        destination_table = f"{self.project}.{self.derived_dataset}.{PROJECTS_TABLE}"

        # check if projects metadata table exists; otherwise it needs to be created
        first_run = True
        try:
            self.bigquery.client.get_table(destination_table)
            first_run = False
        except Exception:
            first_run = True

        project_metadata: List[Dict[str, Any]] = []

        for slug, config in self.projects:
            summaries = config.metrics
            render_summaries = [
                {
                    "metric": summary.metric.name,
                    "statistic": Summary.from_config(summary).statistic.name(),
                }
                for summary in summaries
            ]

            if (
                config.project
                and config.project.end_date
                and config.project.start_date
                and config.project.end_date <= config.project.start_date
            ):
                continue

            metric_groups: Dict[str, List[Any]] = {}
            for metric_group in config.project.metric_groups:
                for metric in metric_group.metrics:
                    if metric.name in metric_groups:
                        metric_groups[metric.name].append(metric_group.name)
                    else:
                        metric_groups[metric.name] = [metric_group.name]

            project_metadata.append(
                {
                    "slug": slug,
                    "dimensions": config.dimensions,
                    "summaries": render_summaries,
                    "config": config.project,
                    "alerts": config.alerts,
                    "metric_groups": metric_groups,
                }
            )

        render_kwargs = {
            "gcp_project": self.project,
            "dataset": self.dataset,
            "derived_dataset": self.derived_dataset,
            "table": PROJECTS_TABLE,
            "projects": project_metadata,
            "first_run": first_run,
        }
        query = self._render_sql(PROJECTS_FILENAME, render_kwargs=render_kwargs)
        self.bigquery.execute(query)

        # Create view
        view_name = PROJECTS_TABLE.split("_")[0]
        view_query = f"""
            CREATE OR REPLACE VIEW `{self.project}.{self.dataset}.{view_name}` AS (
                SELECT *
                FROM `{self.project}.{self.derived_dataset}.{PROJECTS_TABLE}`
            )
        """

        self.bigquery.execute(view_query)
        print("Updated project metadata")