generator.explores.glean_ping_explore
Glean Ping explore type.
1"""Glean Ping explore type.""" 2 3from __future__ import annotations 4 5from pathlib import Path 6from typing import Any, Dict, Iterator, List, Optional 7 8from mozilla_schema_generator.glean_ping import GleanPing 9 10from ..views import GleanPingView, View 11from .ping_explore import PingExplore 12 13 14class GleanPingExplore(PingExplore): 15 """A Glean Ping Table explore.""" 16 17 type: str = "glean_ping_explore" 18 19 def _to_lookml(self, v1_name: Optional[str]) -> List[Dict[str, Any]]: 20 """Generate LookML to represent this explore.""" 21 repo = next((r for r in GleanPing.get_repos() if r["name"] == v1_name)) 22 glean_app = GleanPing(repo) 23 # convert ping description indexes to snake case, as we already have 24 # for the explore name 25 ping_descriptions = { 26 k.replace("-", "_"): v for k, v in glean_app.get_ping_descriptions().items() 27 } 28 # collapse whitespace in the description so the lookml looks a little better 29 ping_description = " ".join(ping_descriptions.get(self.name, "").split()) 30 views_lookml = self.get_view_lookml(self.views["base_view"]) 31 32 # The first view, by convention, is always the base view with the 33 # majority of the dimensions from the top level. 34 base = views_lookml["views"][0] 35 base_name = base["name"] 36 37 joins = [] 38 for view in views_lookml["views"][1:]: 39 if view["name"].startswith("suggest__"): 40 continue 41 view_name = view["name"] 42 metric = "__".join(view["name"].split("__")[1:]) 43 44 if "labeled_counter" in metric: 45 joins.append( 46 { 47 "name": view_name, 48 "relationship": "one_to_many", 49 "sql": ( 50 f"LEFT JOIN UNNEST(${{{base_name}.{metric}}}) AS {view_name} " 51 f"ON ${{{base_name}.document_id}} = ${{{view_name}.document_id}}" 52 ), 53 } 54 ) 55 else: 56 if metric.startswith("metrics__"): 57 continue 58 59 try: 60 # get repeated, nested fields that exist as separate views in lookml 61 base_name, metric = self._get_base_name_and_metric( 62 view_name=view_name, 63 views=[v["name"] for v in views_lookml["views"]], 64 ) 65 metric_name = view_name 66 67 joins.append( 68 { 69 "name": view_name, 70 "relationship": "one_to_many", 71 "sql": ( 72 f"LEFT JOIN UNNEST(${{{base_name}.{metric}}}) AS {metric_name} " 73 ), 74 } 75 ) 76 except Exception: 77 # ignore nested views that cannot be joined on to the base view 78 continue 79 80 base_explore: Dict[str, Any] = { 81 "name": self.name, 82 # list the base explore first by prefixing with a space 83 "view_label": f" {self.name.title()}", 84 "description": f"Explore for the {self.name} ping. {ping_description}", 85 "view_name": self.views["base_view"], 86 "joins": joins, 87 } 88 89 if datagroup := self.get_datagroup(): 90 base_explore["persist_with"] = datagroup 91 92 required_filters = self.get_required_filters("base_view") 93 if len(required_filters) > 0: 94 base_explore["always_filter"] = {"filters": required_filters} 95 96 suggests = [] 97 for view in views_lookml["views"][1:]: 98 if not view["name"].startswith("suggest__"): 99 continue 100 suggests.append({"name": view["name"], "hidden": "yes"}) 101 102 return [base_explore] + suggests 103 104 @staticmethod 105 def from_views(views: List[View]) -> Iterator[PingExplore]: 106 """Generate all possible GleanPingExplores from the views.""" 107 for view in views: 108 if view.view_type == GleanPingView.type: 109 yield GleanPingExplore(view.name, {"base_view": view.name}) 110 111 @staticmethod 112 def from_dict(name: str, defn: dict, views_path: Path) -> GleanPingExplore: 113 """Get an instance of this explore from a name and dictionary definition.""" 114 return GleanPingExplore(name, defn["views"], views_path)
15class GleanPingExplore(PingExplore): 16 """A Glean Ping Table explore.""" 17 18 type: str = "glean_ping_explore" 19 20 def _to_lookml(self, v1_name: Optional[str]) -> List[Dict[str, Any]]: 21 """Generate LookML to represent this explore.""" 22 repo = next((r for r in GleanPing.get_repos() if r["name"] == v1_name)) 23 glean_app = GleanPing(repo) 24 # convert ping description indexes to snake case, as we already have 25 # for the explore name 26 ping_descriptions = { 27 k.replace("-", "_"): v for k, v in glean_app.get_ping_descriptions().items() 28 } 29 # collapse whitespace in the description so the lookml looks a little better 30 ping_description = " ".join(ping_descriptions.get(self.name, "").split()) 31 views_lookml = self.get_view_lookml(self.views["base_view"]) 32 33 # The first view, by convention, is always the base view with the 34 # majority of the dimensions from the top level. 35 base = views_lookml["views"][0] 36 base_name = base["name"] 37 38 joins = [] 39 for view in views_lookml["views"][1:]: 40 if view["name"].startswith("suggest__"): 41 continue 42 view_name = view["name"] 43 metric = "__".join(view["name"].split("__")[1:]) 44 45 if "labeled_counter" in metric: 46 joins.append( 47 { 48 "name": view_name, 49 "relationship": "one_to_many", 50 "sql": ( 51 f"LEFT JOIN UNNEST(${{{base_name}.{metric}}}) AS {view_name} " 52 f"ON ${{{base_name}.document_id}} = ${{{view_name}.document_id}}" 53 ), 54 } 55 ) 56 else: 57 if metric.startswith("metrics__"): 58 continue 59 60 try: 61 # get repeated, nested fields that exist as separate views in lookml 62 base_name, metric = self._get_base_name_and_metric( 63 view_name=view_name, 64 views=[v["name"] for v in views_lookml["views"]], 65 ) 66 metric_name = view_name 67 68 joins.append( 69 { 70 "name": view_name, 71 "relationship": "one_to_many", 72 "sql": ( 73 f"LEFT JOIN UNNEST(${{{base_name}.{metric}}}) AS {metric_name} " 74 ), 75 } 76 ) 77 except Exception: 78 # ignore nested views that cannot be joined on to the base view 79 continue 80 81 base_explore: Dict[str, Any] = { 82 "name": self.name, 83 # list the base explore first by prefixing with a space 84 "view_label": f" {self.name.title()}", 85 "description": f"Explore for the {self.name} ping. {ping_description}", 86 "view_name": self.views["base_view"], 87 "joins": joins, 88 } 89 90 if datagroup := self.get_datagroup(): 91 base_explore["persist_with"] = datagroup 92 93 required_filters = self.get_required_filters("base_view") 94 if len(required_filters) > 0: 95 base_explore["always_filter"] = {"filters": required_filters} 96 97 suggests = [] 98 for view in views_lookml["views"][1:]: 99 if not view["name"].startswith("suggest__"): 100 continue 101 suggests.append({"name": view["name"], "hidden": "yes"}) 102 103 return [base_explore] + suggests 104 105 @staticmethod 106 def from_views(views: List[View]) -> Iterator[PingExplore]: 107 """Generate all possible GleanPingExplores from the views.""" 108 for view in views: 109 if view.view_type == GleanPingView.type: 110 yield GleanPingExplore(view.name, {"base_view": view.name}) 111 112 @staticmethod 113 def from_dict(name: str, defn: dict, views_path: Path) -> GleanPingExplore: 114 """Get an instance of this explore from a name and dictionary definition.""" 115 return GleanPingExplore(name, defn["views"], views_path)
A Glean Ping Table explore.
@staticmethod
def
from_views( views: List[generator.views.view.View]) -> Iterator[generator.explores.ping_explore.PingExplore]:
105 @staticmethod 106 def from_views(views: List[View]) -> Iterator[PingExplore]: 107 """Generate all possible GleanPingExplores from the views.""" 108 for view in views: 109 if view.view_type == GleanPingView.type: 110 yield GleanPingExplore(view.name, {"base_view": view.name})
Generate all possible GleanPingExplores from the views.
112 @staticmethod 113 def from_dict(name: str, defn: dict, views_path: Path) -> GleanPingExplore: 114 """Get an instance of this explore from a name and dictionary definition.""" 115 return GleanPingExplore(name, defn["views"], views_path)
Get an instance of this explore from a name and dictionary definition.