generator.views.operational_monitoring_view

Class to describe an Operational Monitoring View.

 1"""Class to describe an Operational Monitoring View."""
 2
 3from __future__ import annotations
 4
 5from typing import Any, Dict, List, Optional, Union
 6
 7from . import lookml_utils
 8from .ping_view import PingView
 9from .view import ViewDict
10
11ALLOWED_DIMENSIONS = {
12    "branch",
13    "metric",
14    "statistic",
15    "parameter",
16}
17
18
19class OperationalMonitoringView(PingView):
20    """A view on a operational monitoring table."""
21
22    type: str = "operational_monitoring_view"
23
24    def __init__(self, namespace: str, name: str, tables: List[Dict[str, Any]]):
25        """Create instance of a OperationalMonitoringView."""
26        super().__init__(namespace, name, tables)
27        xaxis = "build_id"
28        if "xaxis" in tables[0] and len(tables) > 0:
29            xaxis = tables[0]["xaxis"]
30
31        xaxis_to_sql_mapping = {
32            "build_id": f"PARSE_DATE('%Y%m%d', CAST(${{TABLE}}.{xaxis} AS STRING))",
33            "submission_date": f"${{TABLE}}.{xaxis}",
34        }
35        self.dimensions: List[Dict[str, str]] = [
36            {
37                "name": xaxis,
38                "type": "date",
39                "sql": xaxis_to_sql_mapping[xaxis],
40                "datatype": "date",
41                "convert_tz": "no",
42            }
43        ]
44
45    @classmethod
46    def from_dict(
47        klass, namespace: str, name: str, _dict: ViewDict
48    ) -> OperationalMonitoringView:
49        """Get a OperationalMonitoringView from a dict representation."""
50        return klass(namespace, name, _dict["tables"])
51
52    def to_lookml(self, v1_name: Optional[str], dryrun) -> Dict[str, Any]:
53        """Get this view as LookML."""
54        if len(self.tables) == 0:
55            raise Exception((f"Operational Monitoring view {self.name} has no tables"))
56
57        reference_table = self.tables[0]["table"]
58        all_dimensions = lookml_utils._generate_dimensions(
59            reference_table, dryrun=dryrun
60        )
61
62        filtered_dimensions = [
63            d
64            for d in all_dimensions
65            if d["name"] in ALLOWED_DIMENSIONS
66            or d["name"] in self.tables[0].get("dimensions", {}).keys()
67        ]
68        self.dimensions.extend(filtered_dimensions)
69
70        return {
71            "views": [
72                {
73                    "name": self.name,
74                    "sql_table_name": reference_table,
75                    "dimensions": self.dimensions,
76                    "measures": self.get_measures(
77                        self.dimensions, reference_table, v1_name
78                    ),
79                }
80            ]
81        }
82
83    def get_measures(
84        self, dimensions: List[dict], table: str, v1_name: Optional[str]
85    ) -> List[Dict[str, Union[str, List[Dict[str, str]]]]]:
86        """Get OpMon measures."""
87        return [
88            {"name": "point", "type": "sum", "sql": "${TABLE}.point"},
89            {"name": "upper", "type": "sum", "sql": "${TABLE}.upper"},
90            {"name": "lower", "type": "sum", "sql": "${TABLE}.lower"},
91        ]
ALLOWED_DIMENSIONS = {'branch', 'parameter', 'metric', 'statistic'}
class OperationalMonitoringView(generator.views.ping_view.PingView):
20class OperationalMonitoringView(PingView):
21    """A view on a operational monitoring table."""
22
23    type: str = "operational_monitoring_view"
24
25    def __init__(self, namespace: str, name: str, tables: List[Dict[str, Any]]):
26        """Create instance of a OperationalMonitoringView."""
27        super().__init__(namespace, name, tables)
28        xaxis = "build_id"
29        if "xaxis" in tables[0] and len(tables) > 0:
30            xaxis = tables[0]["xaxis"]
31
32        xaxis_to_sql_mapping = {
33            "build_id": f"PARSE_DATE('%Y%m%d', CAST(${{TABLE}}.{xaxis} AS STRING))",
34            "submission_date": f"${{TABLE}}.{xaxis}",
35        }
36        self.dimensions: List[Dict[str, str]] = [
37            {
38                "name": xaxis,
39                "type": "date",
40                "sql": xaxis_to_sql_mapping[xaxis],
41                "datatype": "date",
42                "convert_tz": "no",
43            }
44        ]
45
46    @classmethod
47    def from_dict(
48        klass, namespace: str, name: str, _dict: ViewDict
49    ) -> OperationalMonitoringView:
50        """Get a OperationalMonitoringView from a dict representation."""
51        return klass(namespace, name, _dict["tables"])
52
53    def to_lookml(self, v1_name: Optional[str], dryrun) -> Dict[str, Any]:
54        """Get this view as LookML."""
55        if len(self.tables) == 0:
56            raise Exception((f"Operational Monitoring view {self.name} has no tables"))
57
58        reference_table = self.tables[0]["table"]
59        all_dimensions = lookml_utils._generate_dimensions(
60            reference_table, dryrun=dryrun
61        )
62
63        filtered_dimensions = [
64            d
65            for d in all_dimensions
66            if d["name"] in ALLOWED_DIMENSIONS
67            or d["name"] in self.tables[0].get("dimensions", {}).keys()
68        ]
69        self.dimensions.extend(filtered_dimensions)
70
71        return {
72            "views": [
73                {
74                    "name": self.name,
75                    "sql_table_name": reference_table,
76                    "dimensions": self.dimensions,
77                    "measures": self.get_measures(
78                        self.dimensions, reference_table, v1_name
79                    ),
80                }
81            ]
82        }
83
84    def get_measures(
85        self, dimensions: List[dict], table: str, v1_name: Optional[str]
86    ) -> List[Dict[str, Union[str, List[Dict[str, str]]]]]:
87        """Get OpMon measures."""
88        return [
89            {"name": "point", "type": "sum", "sql": "${TABLE}.point"},
90            {"name": "upper", "type": "sum", "sql": "${TABLE}.upper"},
91            {"name": "lower", "type": "sum", "sql": "${TABLE}.lower"},
92        ]

A view on a operational monitoring table.

OperationalMonitoringView(namespace: str, name: str, tables: List[Dict[str, Any]])
25    def __init__(self, namespace: str, name: str, tables: List[Dict[str, Any]]):
26        """Create instance of a OperationalMonitoringView."""
27        super().__init__(namespace, name, tables)
28        xaxis = "build_id"
29        if "xaxis" in tables[0] and len(tables) > 0:
30            xaxis = tables[0]["xaxis"]
31
32        xaxis_to_sql_mapping = {
33            "build_id": f"PARSE_DATE('%Y%m%d', CAST(${{TABLE}}.{xaxis} AS STRING))",
34            "submission_date": f"${{TABLE}}.{xaxis}",
35        }
36        self.dimensions: List[Dict[str, str]] = [
37            {
38                "name": xaxis,
39                "type": "date",
40                "sql": xaxis_to_sql_mapping[xaxis],
41                "datatype": "date",
42                "convert_tz": "no",
43            }
44        ]

Create instance of a OperationalMonitoringView.

type: str = 'operational_monitoring_view'
dimensions: List[Dict[str, str]]
@classmethod
def from_dict( klass, namespace: str, name: str, _dict: generator.views.view.ViewDict) -> OperationalMonitoringView:
46    @classmethod
47    def from_dict(
48        klass, namespace: str, name: str, _dict: ViewDict
49    ) -> OperationalMonitoringView:
50        """Get a OperationalMonitoringView from a dict representation."""
51        return klass(namespace, name, _dict["tables"])

Get a OperationalMonitoringView from a dict representation.

def to_lookml(self, v1_name: Optional[str], dryrun) -> Dict[str, Any]:
53    def to_lookml(self, v1_name: Optional[str], dryrun) -> Dict[str, Any]:
54        """Get this view as LookML."""
55        if len(self.tables) == 0:
56            raise Exception((f"Operational Monitoring view {self.name} has no tables"))
57
58        reference_table = self.tables[0]["table"]
59        all_dimensions = lookml_utils._generate_dimensions(
60            reference_table, dryrun=dryrun
61        )
62
63        filtered_dimensions = [
64            d
65            for d in all_dimensions
66            if d["name"] in ALLOWED_DIMENSIONS
67            or d["name"] in self.tables[0].get("dimensions", {}).keys()
68        ]
69        self.dimensions.extend(filtered_dimensions)
70
71        return {
72            "views": [
73                {
74                    "name": self.name,
75                    "sql_table_name": reference_table,
76                    "dimensions": self.dimensions,
77                    "measures": self.get_measures(
78                        self.dimensions, reference_table, v1_name
79                    ),
80                }
81            ]
82        }

Get this view as LookML.

def get_measures( self, dimensions: List[dict], table: str, v1_name: Optional[str]) -> List[Dict[str, Union[str, List[Dict[str, str]]]]]:
84    def get_measures(
85        self, dimensions: List[dict], table: str, v1_name: Optional[str]
86    ) -> List[Dict[str, Union[str, List[Dict[str, str]]]]]:
87        """Get OpMon measures."""
88        return [
89            {"name": "point", "type": "sum", "sql": "${TABLE}.point"},
90            {"name": "upper", "type": "sum", "sql": "${TABLE}.upper"},
91            {"name": "lower", "type": "sum", "sql": "${TABLE}.lower"},
92        ]

Get OpMon measures.