generator.views.view

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