generator.views.table_view

  1"""Class to describe a Table View."""
  2
  3from __future__ import annotations
  4
  5from collections import defaultdict
  6from itertools import filterfalse
  7from typing import Any, Dict, Iterator, List, Optional, Set
  8
  9from click import ClickException
 10
 11from . import lookml_utils
 12from .view import OMIT_VIEWS, View, ViewDict
 13
 14
 15class TableView(View):
 16    """A view on any table."""
 17
 18    type: str = "table_view"
 19    measures: Optional[Dict[str, Dict[str, Any]]]
 20
 21    def __init__(
 22        self,
 23        namespace: str,
 24        name: str,
 25        tables: List[Dict[str, str]],
 26        measures: Optional[Dict[str, Dict[str, Any]]] = None,
 27    ):
 28        """Create instance of a TableView."""
 29        super().__init__(namespace, name, TableView.type, tables)
 30        self.measures = measures
 31
 32    @classmethod
 33    def from_db_views(
 34        klass,
 35        namespace: str,
 36        is_glean: bool,
 37        channels: List[Dict[str, str]],
 38        db_views: dict,
 39    ) -> Iterator[TableView]:
 40        """Get Looker views for a namespace."""
 41        view_tables: Dict[str, Dict[str, Dict[str, str]]] = defaultdict(dict)
 42        for channel in channels:
 43            dataset = channel["dataset"]
 44
 45            for view_id, references in db_views[dataset].items():
 46                if view_id in OMIT_VIEWS:
 47                    continue
 48
 49                table_id = f"mozdata.{dataset}.{view_id}"
 50                table: Dict[str, str] = {"table": table_id}
 51                if "channel" in channel:
 52                    table["channel"] = channel["channel"]
 53
 54                view_tables[view_id][table_id] = table
 55
 56        for view_id, tables_by_id in view_tables.items():
 57            yield TableView(namespace, f"{view_id}_table", list(tables_by_id.values()))
 58
 59    @classmethod
 60    def from_dict(klass, namespace: str, name: str, _dict: ViewDict) -> TableView:
 61        """Get a view from a name and dict definition."""
 62        return TableView(namespace, name, _dict["tables"], _dict.get("measures"))
 63
 64    def to_lookml(self, v1_name: Optional[str], dryrun) -> Dict[str, Any]:
 65        """Generate LookML for this view."""
 66        view_defn: Dict[str, Any] = {"name": self.name}
 67
 68        # use schema for the table where channel=="release" or the first one
 69        table = next(
 70            (table for table in self.tables if table.get("channel") == "release"),
 71            self.tables[0],
 72        )["table"]
 73
 74        # add dimensions and dimension groups
 75        dimensions = lookml_utils._generate_dimensions(table, dryrun=dryrun)
 76        view_defn["dimensions"] = list(
 77            filterfalse(lookml_utils._is_dimension_group, dimensions)
 78        )
 79        view_defn["dimension_groups"] = list(
 80            filter(lookml_utils._is_dimension_group, dimensions)
 81        )
 82
 83        # add tag "time_partitioning_field"
 84        time_partitioning_fields: Set[str] = set(
 85            # filter out falsy values
 86            filter(
 87                None, (table.get("time_partitioning_field") for table in self.tables)
 88            )
 89        )
 90        if len(time_partitioning_fields) > 1:
 91            raise ClickException(f"Multiple time_partitioning_fields for {self.name!r}")
 92        elif len(time_partitioning_fields) == 1:
 93            field_name = time_partitioning_fields.pop()
 94            sql = f"${{TABLE}}.{field_name}"
 95            for group_defn in view_defn["dimension_groups"]:
 96                if group_defn["sql"] == sql:
 97                    if "tags" not in group_defn:
 98                        group_defn["tags"] = []
 99                    group_defn["tags"].append("time_partitioning_field")
100                    break
101            else:
102                raise ClickException(
103                    f"time_partitioning_field {field_name!r} not found in {self.name!r}"
104                )
105
106        [project, dataset, table_id] = table.split(".")
107        table_schema = dryrun.create(
108            project=project,
109            dataset=dataset,
110            table=table_id,
111        ).get_table_schema()
112        nested_views = lookml_utils._generate_nested_dimension_views(
113            table_schema, self.name
114        )
115
116        if self.measures:
117            view_defn["measures"] = [
118                {"name": measure_name, **measure_parameters}
119                for measure_name, measure_parameters in self.measures.items()
120            ]
121
122        # parameterize table name
123        if len(self.tables) > 1:
124            view_defn["parameters"] = [
125                {
126                    "name": "channel",
127                    "type": "unquoted",
128                    "default_value": table,
129                    "allowed_values": [
130                        {
131                            "label": _table["channel"].title(),
132                            "value": _table["table"],
133                        }
134                        for _table in self.tables
135                    ],
136                }
137            ]
138            view_defn["sql_table_name"] = "`{% parameter channel %}`"
139        else:
140            view_defn["sql_table_name"] = f"`{table}`"
141
142        return {"views": [view_defn] + nested_views}