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)
class GleanPingExplore(generator.explores.ping_explore.PingExplore):
 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.

type: str = 'glean_ping_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.

@staticmethod
def from_dict( name: str, defn: dict, views_path: pathlib.Path) -> GleanPingExplore:
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.