generator.views.view

  1"""Generic class to describe Looker views."""
  2
  3from __future__ import annotations
  4
  5from typing import Any, Dict, Iterator, List, NotRequired, Optional, Set, TypedDict
  6
  7from click import ClickException
  8
  9OMIT_VIEWS: Set[str] = set()
 10
 11
 12class ViewDict(TypedDict):
 13    """Represent a view definition."""
 14
 15    type: str
 16    tables: List[Dict[str, str]]
 17    measures: NotRequired[Dict[str, Dict[str, Any]]]
 18
 19
 20class View(object):
 21    """A generic Looker View."""
 22
 23    name: str
 24    view_type: str
 25    tables: List[Dict[str, Any]]
 26    namespace: str
 27
 28    def __init__(
 29        self,
 30        namespace: str,
 31        name: str,
 32        view_type: str,
 33        tables: List[Dict[str, Any]],
 34        **kwargs,
 35    ):
 36        """Create an instance of a view."""
 37        self.namespace = namespace
 38        self.tables = tables
 39        self.name = name
 40        self.view_type = view_type
 41
 42    @classmethod
 43    def from_db_views(
 44        klass,
 45        namespace: str,
 46        is_glean: bool,
 47        channels: List[Dict[str, str]],
 48        db_views: dict,
 49    ) -> Iterator[View]:
 50        """Get Looker views from app."""
 51        raise NotImplementedError("Only implemented in subclass.")
 52
 53    @classmethod
 54    def from_dict(klass, namespace: str, name: str, _dict: ViewDict) -> View:
 55        """Get a view from a name and dict definition."""
 56        raise NotImplementedError("Only implemented in subclass.")
 57
 58    def get_type(self) -> str:
 59        """Get the type of this view."""
 60        return self.view_type
 61
 62    def as_dict(self) -> dict:
 63        """Get this view as a dictionary."""
 64        return {
 65            "type": self.view_type,
 66            "tables": self.tables,
 67        }
 68
 69    def __str__(self):
 70        """Stringify."""
 71        return f"name: {self.name}, type: {self.type}, table: {self.tables}, namespace: {self.namespace}"
 72
 73    def __eq__(self, other) -> bool:
 74        """Check for equality with other View."""
 75
 76        def comparable_dict(d):
 77            return {tuple(sorted([(k, str(v)) for k, v in t.items()])) for t in d}
 78
 79        if isinstance(other, View):
 80            return (
 81                self.name == other.name
 82                and self.view_type == other.view_type
 83                and comparable_dict(self.tables) == comparable_dict(other.tables)
 84                and self.namespace == other.namespace
 85            )
 86        return False
 87
 88    def get_dimensions(
 89        self, table, v1_name: Optional[str], dryrun
 90    ) -> List[Dict[str, Any]]:
 91        """Get the set of dimensions for this view."""
 92        raise NotImplementedError("Only implemented in subclass.")
 93
 94    def to_lookml(self, v1_name: Optional[str], dryrun) -> Dict[str, Any]:
 95        """
 96        Generate Lookml for this view.
 97
 98        View instances can generate more than one Looker view,
 99        for e.g. nested fields and joins, so this returns
100        a list.
101        """
102        raise NotImplementedError("Only implemented in subclass.")
103
104    def get_client_id(self, dimensions: List[dict], table: str) -> Optional[str]:
105        """Return the first field that looks like a client identifier."""
106        client_id_fields = self.select_dimension(
107            {"client_id", "client_info__client_id", "context_id"},
108            dimensions,
109            table,
110        )
111        # Some pings purposely disinclude client_ids, e.g. firefox installer
112        return client_id_fields["name"] if client_id_fields else None
113
114    def get_document_id(self, dimensions: List[dict], table: str) -> Optional[str]:
115        """Return the first field that looks like a document_id."""
116        document_id = self.select_dimension("document_id", dimensions, table)
117        return document_id["name"] if document_id else None
118
119    def select_dimension(
120        self,
121        dimension_names: str | set[str],
122        dimensions: List[dict],
123        table: str,
124    ) -> Optional[dict[str, str]]:
125        """
126        Return the first field that matches dimension name.
127
128        Throws if the query set is greater than one and more than one item is selected.
129        """
130        if isinstance(dimension_names, str):
131            dimension_names = {dimension_names}
132        selected = [d for d in dimensions if d["name"] in dimension_names]
133        if selected:
134            # there should only be one dimension selected from the set
135            # if there are multiple options in the dimention_names set.
136            if len(dimension_names) > 1 and len(selected) > 1:
137                raise ClickException(
138                    f"Duplicate {'/'.join(dimension_names)} dimension in {table!r}"
139                )
140            return selected[0]
141        return None