generator.explores.table_explore

 1"""Table explore type."""
 2
 3from __future__ import annotations
 4
 5from pathlib import Path
 6from typing import Any, Dict, Iterator, List, Optional
 7
 8from ..views import TableView, View
 9from . import Explore
10
11ALLOWED_VIEWS = {"events_stream_table"}
12
13
14class TableExplore(Explore):
15    """A table explore."""
16
17    type: str = "table_explore"
18
19    def _to_lookml(self, v1_name: Optional[str]) -> List[Dict[str, Any]]:
20        """Generate LookML to represent this explore."""
21        explore_lookml: Dict[str, Any] = {
22            "name": self.name,
23            "view_name": self.views["base_view"],
24            "joins": self.get_unnested_fields_joins_lookml(),
25        }
26        if required_filters := self.get_required_filters("base_view"):
27            explore_lookml["always_filter"] = {
28                "filters": required_filters,
29            }
30
31        if datagroup := self.get_datagroup():
32            explore_lookml["persist_with"] = datagroup
33
34        return [explore_lookml]
35
36    @staticmethod
37    def from_views(views: List[View]) -> Iterator[TableExplore]:
38        """Don't generate all possible TableExplores from the views, only generate for ALLOWED_VIEWS."""
39        for view in views:
40            if view.view_type == TableView.type:
41                if view.name in ALLOWED_VIEWS:
42                    yield TableExplore(view.name, {"base_view": view.name})
43
44    @staticmethod
45    def from_dict(name: str, defn: dict, views_path: Path) -> TableExplore:
46        """Get an instance of this explore from a name and dictionary definition."""
47        return TableExplore(name, defn["views"], views_path)
48
49    def get_datagroup(self) -> Optional[str]:
50        """
51        Return the name of the associated datagroup.
52
53        Return `None` if there is no datagroup for this explore.
54        """
55        if self.views_path and (self.views_path.parent / "datagroups").exists():
56            datagroups_path = self.views_path.parent / "datagroups"
57            datagroup_file = (
58                datagroups_path
59                / f'{self.views["base_view"]}_last_updated.datagroup.lkml'
60            )
61            if datagroup_file.exists():
62                return f'{self.views["base_view"]}_last_updated'
63        return None