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 = { 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 "always_filter": { 87 "filters": self.get_required_filters("base_view"), 88 }, 89 "joins": joins, 90 } 91 92 suggests = [] 93 for view in views_lookml["views"][1:]: 94 if not view["name"].startswith("suggest__"): 95 continue 96 suggests.append({"name": view["name"], "hidden": "yes"}) 97 98 return [base_explore] + suggests 99 100 @staticmethod 101 def from_views(views: List[View]) -> Iterator[PingExplore]: 102 """Generate all possible GleanPingExplores from the views.""" 103 for view in views: 104 if view.view_type == GleanPingView.type: 105 yield GleanPingExplore(view.name, {"base_view": view.name}) 106 107 @staticmethod 108 def from_dict(name: str, defn: dict, views_path: Path) -> GleanPingExplore: 109 """Get an instance of this explore from a name and dictionary definition.""" 110 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 = { 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 "always_filter": { 88 "filters": self.get_required_filters("base_view"), 89 }, 90 "joins": joins, 91 } 92 93 suggests = [] 94 for view in views_lookml["views"][1:]: 95 if not view["name"].startswith("suggest__"): 96 continue 97 suggests.append({"name": view["name"], "hidden": "yes"}) 98 99 return [base_explore] + suggests 100 101 @staticmethod 102 def from_views(views: List[View]) -> Iterator[PingExplore]: 103 """Generate all possible GleanPingExplores from the views.""" 104 for view in views: 105 if view.view_type == GleanPingView.type: 106 yield GleanPingExplore(view.name, {"base_view": view.name}) 107 108 @staticmethod 109 def from_dict(name: str, defn: dict, views_path: Path) -> GleanPingExplore: 110 """Get an instance of this explore from a name and dictionary definition.""" 111 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]:
101 @staticmethod 102 def from_views(views: List[View]) -> Iterator[PingExplore]: 103 """Generate all possible GleanPingExplores from the views.""" 104 for view in views: 105 if view.view_type == GleanPingView.type: 106 yield GleanPingExplore(view.name, {"base_view": view.name})
Generate all possible GleanPingExplores from the views.
108 @staticmethod 109 def from_dict(name: str, defn: dict, views_path: Path) -> GleanPingExplore: 110 """Get an instance of this explore from a name and dictionary definition.""" 111 return GleanPingExplore(name, defn["views"], views_path)
Get an instance of this explore from a name and dictionary definition.