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', 'metric', 'statistic', 'parameter'}
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.
@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.