def _get_opmon()

in generator/namespaces.py [0:0]
82 lines of code
11 McCabe index (conditional complexity)

def _get_opmon(bq_client: bigquery.Client, namespaces: Dict[str, Any]):
    om_content: Dict[str, Any] = {"views": {}, "explores": {}, "dashboards": {}}
    # get operational monitoring namespace information

    opmon_namespace = namespaces["operational_monitoring"]
    views = opmon_namespace.get("views")

    if views is None:
        print("No views defined for operational monitoring")
        return {}

    projects_view = views.get("projects")

    if projects_view is None:
        print("No projects view defined for operational monitoring")
        return {}

    projects_table = projects_view["tables"][0]["table"]
    projects = operational_monitoring_utils.get_active_projects(
        bq_client, project_table=projects_table
    )

    # Iterating over all defined operational monitoring projects
    for project in projects:
        table_prefix = _normalize_slug(project["slug"])
        project_name = lookml_utils.slug_to_title(
            re.sub("[^0-9a-zA-Z_]+", "_", "_".join(project["name"].lower().split(" ")))
        )
        branches = project.get("branches", ["enabled", "disabled"])

        # append view and explore for data type
        table = f"{PROD_PROJECT}.{OPMON_DATASET}.{table_prefix}_statistics"
        dimensions = operational_monitoring_utils.get_dimension_defaults(
            bq_client, table, project["dimensions"]
        )
        om_content["views"][table_prefix] = {
            "type": "operational_monitoring_view",
            "tables": [
                {
                    "table": table,
                    "xaxis": project["xaxis"],
                    "dimensions": dimensions,
                }
            ],
        }
        om_content["explores"][table_prefix] = {
            "type": "operational_monitoring_explore",
            "views": {"base_view": f"{table_prefix}"},
            "branches": branches,
            "xaxis": project["xaxis"],
            "dimensions": dimensions,
            "summaries": project["summaries"],
        }

        if "alerting" in project and project["alerting"]:
            # create an alerting view if available
            om_content["views"][f"{table_prefix}_alerts"] = {
                "type": "operational_monitoring_alerting_view",
                "tables": [
                    {
                        "table": f"{PROD_PROJECT}.{OPMON_DATASET}.{table_prefix}_alerts",
                    }
                ],
            }
            om_content["explores"][f"{table_prefix}_alerts"] = {
                "type": "operational_monitoring_alerting_explore",
                "views": {"base_view": f"{table_prefix}_alerts"},
            }

        om_content["dashboards"][table_prefix] = {
            "type": "operational_monitoring_dashboard",
            "title": project_name,
            "tables": [
                {
                    "explore": f"{table_prefix}",
                    "table": f"{PROD_PROJECT}.{OPMON_DATASET}.{table_prefix}_statistics",
                    "branches": branches,
                    "xaxis": project["xaxis"],
                    "compact_visualization": project.get(
                        "compact_visualization", False
                    ),
                    "dimensions": dimensions,
                    "group_by_dimension": project.get("group_by_dimension", None),
                    "summaries": project["summaries"],
                }
            ],
        }

        if "alerting" in project and project["alerting"]:
            om_content["dashboards"][table_prefix]["tables"].append(
                {
                    "explore": f"{table_prefix}_alerts",
                    "table": f"{PROD_PROJECT}.{OPMON_DATASET}.{table_prefix}_alerts",
                }
            )

    return om_content