generator.explores.events_explore
An explore for Events Views.
1"""An explore for Events Views.""" 2 3from __future__ import annotations 4 5from pathlib import Path 6from typing import Any, Dict, Iterator, List, Optional 7 8from ..views import EventsView, View 9from .explore import Explore 10 11 12class EventsExplore(Explore): 13 """An Events Explore, from any unnested events table.""" 14 15 type: str = "events_explore" 16 17 def get_required_filters(self, view_name: str) -> List[Dict[str, str]]: 18 """Get required filters for this view. 19 20 Override the default to use 7 days instead of 28 days to avoid 21 "Query Exceeds Data Limit" errors for large event datasets. 22 """ 23 filters = [] 24 view = self.views[view_name] 25 26 # Add a default filter on channel, if it's present in the view 27 default_channel = self._get_default_channel(view) 28 if default_channel is not None: 29 filters.append({"channel": default_channel}) 30 31 # Add submission filter with 7 days instead of the default 28 days 32 if time_partitioning_group := self.get_view_time_partitioning_group(view): 33 filters.append({f"{time_partitioning_group}_date": "7 days"}) 34 35 return filters 36 37 @staticmethod 38 def from_views(views: List[View]) -> Iterator[EventsExplore]: 39 """Where possible, generate EventsExplores for Views.""" 40 for view in views: 41 if isinstance(view, EventsView): 42 yield EventsExplore( 43 view.name, 44 { 45 "base_view": "events", 46 "extended_view": view.tables[0]["events_table_view"], 47 }, 48 ) 49 50 @staticmethod 51 def from_dict(name: str, defn: dict, views_path: Path) -> EventsExplore: 52 """Get an instance of this explore from a dictionary definition.""" 53 return EventsExplore(name, defn["views"], views_path) 54 55 def _to_lookml(self, v1_name: Optional[str]) -> List[Dict[str, Any]]: 56 name = self.name 57 if not name.endswith("_counts"): 58 name = "event_counts" 59 60 lookml: Dict[str, Any] = { 61 "name": name, 62 "view_name": self.views["base_view"], 63 "description": "Event counts over time.", 64 "joins": self.get_unnested_fields_joins_lookml(), 65 } 66 if required_filters := self.get_required_filters("extended_view"): 67 lookml["always_filter"] = {"filters": required_filters} 68 if time_partitioning_group := self.get_view_time_partitioning_group( 69 self.views["extended_view"] 70 ): 71 date_dimension = f"{time_partitioning_group}_date" 72 lookml["queries"] = [ 73 { 74 "description": "Event counts from all events over the past two weeks.", 75 "dimensions": [date_dimension], 76 "measures": ["event_count"], 77 "filters": [ 78 {date_dimension: "14 days"}, 79 ], 80 "name": "all_event_counts", 81 }, 82 ] 83 84 if datagroup := self.get_datagroup(): 85 lookml["persist_with"] = datagroup 86 87 return [lookml]
13class EventsExplore(Explore): 14 """An Events Explore, from any unnested events table.""" 15 16 type: str = "events_explore" 17 18 def get_required_filters(self, view_name: str) -> List[Dict[str, str]]: 19 """Get required filters for this view. 20 21 Override the default to use 7 days instead of 28 days to avoid 22 "Query Exceeds Data Limit" errors for large event datasets. 23 """ 24 filters = [] 25 view = self.views[view_name] 26 27 # Add a default filter on channel, if it's present in the view 28 default_channel = self._get_default_channel(view) 29 if default_channel is not None: 30 filters.append({"channel": default_channel}) 31 32 # Add submission filter with 7 days instead of the default 28 days 33 if time_partitioning_group := self.get_view_time_partitioning_group(view): 34 filters.append({f"{time_partitioning_group}_date": "7 days"}) 35 36 return filters 37 38 @staticmethod 39 def from_views(views: List[View]) -> Iterator[EventsExplore]: 40 """Where possible, generate EventsExplores for Views.""" 41 for view in views: 42 if isinstance(view, EventsView): 43 yield EventsExplore( 44 view.name, 45 { 46 "base_view": "events", 47 "extended_view": view.tables[0]["events_table_view"], 48 }, 49 ) 50 51 @staticmethod 52 def from_dict(name: str, defn: dict, views_path: Path) -> EventsExplore: 53 """Get an instance of this explore from a dictionary definition.""" 54 return EventsExplore(name, defn["views"], views_path) 55 56 def _to_lookml(self, v1_name: Optional[str]) -> List[Dict[str, Any]]: 57 name = self.name 58 if not name.endswith("_counts"): 59 name = "event_counts" 60 61 lookml: Dict[str, Any] = { 62 "name": name, 63 "view_name": self.views["base_view"], 64 "description": "Event counts over time.", 65 "joins": self.get_unnested_fields_joins_lookml(), 66 } 67 if required_filters := self.get_required_filters("extended_view"): 68 lookml["always_filter"] = {"filters": required_filters} 69 if time_partitioning_group := self.get_view_time_partitioning_group( 70 self.views["extended_view"] 71 ): 72 date_dimension = f"{time_partitioning_group}_date" 73 lookml["queries"] = [ 74 { 75 "description": "Event counts from all events over the past two weeks.", 76 "dimensions": [date_dimension], 77 "measures": ["event_count"], 78 "filters": [ 79 {date_dimension: "14 days"}, 80 ], 81 "name": "all_event_counts", 82 }, 83 ] 84 85 if datagroup := self.get_datagroup(): 86 lookml["persist_with"] = datagroup 87 88 return [lookml]
An Events Explore, from any unnested events table.
def
get_required_filters(self, view_name: str) -> List[Dict[str, str]]:
18 def get_required_filters(self, view_name: str) -> List[Dict[str, str]]: 19 """Get required filters for this view. 20 21 Override the default to use 7 days instead of 28 days to avoid 22 "Query Exceeds Data Limit" errors for large event datasets. 23 """ 24 filters = [] 25 view = self.views[view_name] 26 27 # Add a default filter on channel, if it's present in the view 28 default_channel = self._get_default_channel(view) 29 if default_channel is not None: 30 filters.append({"channel": default_channel}) 31 32 # Add submission filter with 7 days instead of the default 28 days 33 if time_partitioning_group := self.get_view_time_partitioning_group(view): 34 filters.append({f"{time_partitioning_group}_date": "7 days"}) 35 36 return filters
Get required filters for this view.
Override the default to use 7 days instead of 28 days to avoid "Query Exceeds Data Limit" errors for large event datasets.
38 @staticmethod 39 def from_views(views: List[View]) -> Iterator[EventsExplore]: 40 """Where possible, generate EventsExplores for Views.""" 41 for view in views: 42 if isinstance(view, EventsView): 43 yield EventsExplore( 44 view.name, 45 { 46 "base_view": "events", 47 "extended_view": view.tables[0]["events_table_view"], 48 }, 49 )
Where possible, generate EventsExplores for Views.