generator.operational_monitoring_utils

 1"""Utils for operational monitoring."""
 2
 3from multiprocessing.pool import ThreadPool
 4from typing import Any, Dict, List, Optional, Tuple
 5
 6from google.api_core import exceptions
 7from google.cloud import bigquery
 8
 9from .views import lookml_utils
10
11
12def _default_helper(
13    bq_client: bigquery.Client, table: str, dimension: str
14) -> Tuple[Optional[str], dict]:
15    query_job = bq_client.query(
16        f"""
17            SELECT DISTINCT {dimension} AS option, COUNT(*)
18            FROM {table}
19            WHERE {dimension} IS NOT NULL
20            GROUP BY 1
21            ORDER BY 2 DESC
22            LIMIT 10
23        """
24    )
25
26    dimension_options = list(query_job.result())
27
28    if len(dimension_options) > 0:
29        return dimension, {
30            "default": dimension_options[0]["option"],
31            "options": [d["option"] for d in dimension_options],
32        }
33    return None, {}
34
35
36def get_dimension_defaults(
37    bq_client: bigquery.Client, table: str, dimensions: List[str]
38) -> Dict[str, Any]:
39    """
40    Find default values for certain dimensions.
41
42    For a given Operational Monitoring dimension, find its default (most common)
43    value and its top 10 most common to be used as dropdown options.
44    """
45    with ThreadPool(4) as pool:
46        return {
47            key: value
48            for key, value in pool.starmap(
49                _default_helper,
50                [[bq_client, table, dimension] for dimension in dimensions],
51            )
52            if key is not None
53        }
54
55
56def get_xaxis_val(bq_client: bigquery.Client, table: str) -> str:
57    """
58    Return whether the x-axis should be build_id or submission_date.
59
60    This is based on which one is found in the table provided.
61    """
62    all_dimensions = lookml_utils._generate_dimensions(bq_client, table)
63    return (
64        "build_id"
65        if "build_id" in {dimension["name"] for dimension in all_dimensions}
66        else "submission_date"
67    )
68
69
70def get_active_projects(
71    bq_client: bigquery.Client, project_table: str
72) -> List[Dict[str, Any]]:
73    """Select all operational monitoring projects."""
74    try:
75        query_job = bq_client.query(
76            f"""
77                SELECT *
78                FROM `{project_table}`
79                WHERE
80                    DATE_ADD(end_date, INTERVAL 90 DAY) > CURRENT_DATE() OR
81                    end_date IS NULL
82            """
83        )
84
85        projects = [dict(row) for row in query_job.result()]
86    except exceptions.Forbidden:
87        projects = []
88    return projects