From 3161f37a026f1a56c76cb221291c488c11ed1b43 Mon Sep 17 00:00:00 2001 From: Maxime Armstrong Date: Fri, 27 Dec 2024 14:18:49 -0500 Subject: [PATCH] [dagster-looker] Move contextual data from DagsterLookerApiTranslator to LookerApiTranslatorStructureData --- .../looker/customize-looker-assets.py | 2 +- .../dagster_looker/api/assets.py | 16 +++-- .../api/dagster_looker_api_translator.py | 65 +++++++++++++------ .../dagster_looker/api/resource.py | 27 +++++--- .../api/test_build_defs.py | 4 +- 5 files changed, 76 insertions(+), 38 deletions(-) diff --git a/examples/docs_snippets/docs_snippets/integrations/looker/customize-looker-assets.py b/examples/docs_snippets/docs_snippets/integrations/looker/customize-looker-assets.py index 57adf4935ff8f..981297d84a1a2 100644 --- a/examples/docs_snippets/docs_snippets/integrations/looker/customize-looker-assets.py +++ b/examples/docs_snippets/docs_snippets/integrations/looker/customize-looker-assets.py @@ -18,7 +18,7 @@ class CustomDagsterLookerApiTranslator(DagsterLookerApiTranslator): def get_asset_spec(self, looker_structure: LookerStructureData) -> dg.AssetSpec: # We create the default asset spec using super() - default_spec = super().get_asset_spec(looker_structure) + default_spec = super().get_asset_spec(looker_structure) # type: ignore # We customize the team owner tag for all assets, # and we customize the asset key prefix only for dashboards. return default_spec.replace_attributes( diff --git a/python_modules/libraries/dagster-looker/dagster_looker/api/assets.py b/python_modules/libraries/dagster-looker/dagster_looker/api/assets.py index 505224d255106..299cd33ccb380 100644 --- a/python_modules/libraries/dagster-looker/dagster_looker/api/assets.py +++ b/python_modules/libraries/dagster-looker/dagster_looker/api/assets.py @@ -5,6 +5,7 @@ from dagster_looker.api.dagster_looker_api_translator import ( DagsterLookerApiTranslator, + LookerApiTranslatorStructureData, LookerStructureData, LookerStructureType, LookmlView, @@ -32,19 +33,22 @@ def build_looker_pdt_assets_definitions( Returns: AssetsDefinition: The AssetsDefinitions of the executable assets for the given the list of refreshable PDTs. """ - translator = dagster_looker_translator(None) + translator = dagster_looker_translator() result = [] for request_start_pdt_build in request_start_pdt_builds: @multi_asset( specs=[ translator.get_asset_spec( - LookerStructureData( - structure_type=LookerStructureType.VIEW, - data=LookmlView( - view_name=request_start_pdt_build.view_name, - sql_table_name=None, + LookerApiTranslatorStructureData( + structure_data=LookerStructureData( + structure_type=LookerStructureType.VIEW, + data=LookmlView( + view_name=request_start_pdt_build.view_name, + sql_table_name=None, + ), ), + instance_data=None, ) ) ], diff --git a/python_modules/libraries/dagster-looker/dagster_looker/api/dagster_looker_api_translator.py b/python_modules/libraries/dagster-looker/dagster_looker/api/dagster_looker_api_translator.py index ab2476daa4e87..03733e9bb99d5 100644 --- a/python_modules/libraries/dagster-looker/dagster_looker/api/dagster_looker_api_translator.py +++ b/python_modules/libraries/dagster-looker/dagster_looker/api/dagster_looker_api_translator.py @@ -107,22 +107,37 @@ class LookerStructureData: base_url: Optional[str] = None -class DagsterLookerApiTranslator: - def __init__(self, looker_instance_data: Optional[LookerInstanceData]): - self._looker_instance_data = looker_instance_data +@record +class LookerApiTranslatorStructureData: + """A record representing a structure in Looker and the Looker instance data. + Includes the content's type and data as returned from the API. + """ + + structure_data: "LookerStructureData" + instance_data: Optional["LookerInstanceData"] + + @property + def structure_type(self) -> LookerStructureType: + return self.structure_data.structure_type + + @property + def data(self) -> Union[LookmlView, LookmlModelExplore, DashboardFilter, Dashboard]: + return self.structure_data.data @property - def instance_data(self) -> Optional[LookerInstanceData]: - return self._looker_instance_data + def base_url(self) -> Optional[str]: + return self.structure_data.base_url + +class DagsterLookerApiTranslator: @deprecated( breaking_version="1.10", additional_warn_text="Use `DagsterLookerApiTranslator.get_asset_spec().key` instead", ) - def get_view_asset_key(self, looker_structure: LookerStructureData) -> AssetKey: + def get_view_asset_key(self, looker_structure: LookerApiTranslatorStructureData) -> AssetKey: return self.get_asset_spec(looker_structure).key - def get_view_asset_spec(self, looker_structure: LookerStructureData) -> AssetSpec: + def get_view_asset_spec(self, looker_structure: LookerApiTranslatorStructureData) -> AssetSpec: lookml_view = check.inst(looker_structure.data, LookmlView) return AssetSpec( key=AssetKey(["view", lookml_view.view_name]), @@ -132,10 +147,12 @@ def get_view_asset_spec(self, looker_structure: LookerStructureData) -> AssetSpe breaking_version="1.10", additional_warn_text="Use `DagsterLookerApiTranslator.get_asset_spec().key` instead", ) - def get_explore_asset_key(self, looker_structure: LookerStructureData) -> AssetKey: + def get_explore_asset_key(self, looker_structure: LookerApiTranslatorStructureData) -> AssetKey: return self.get_explore_asset_spec(looker_structure).key - def get_explore_asset_spec(self, looker_structure: LookerStructureData) -> AssetSpec: + def get_explore_asset_spec( + self, looker_structure: LookerApiTranslatorStructureData + ) -> AssetSpec: lookml_explore = check.inst(looker_structure.data, (LookmlModelExplore, DashboardFilter)) if isinstance(lookml_explore, LookmlModelExplore): @@ -157,8 +174,11 @@ def get_explore_asset_spec(self, looker_structure: LookerStructureData) -> Asset deps=list( { self.get_asset_spec( - LookerStructureData( - structure_type=LookerStructureType.VIEW, data=lookml_view + LookerApiTranslatorStructureData( + structure_data=LookerStructureData( + structure_type=LookerStructureType.VIEW, data=lookml_view + ), + instance_data=looker_structure.instance_data, ) ).key for lookml_view in [explore_base_view, *explore_join_views] @@ -185,23 +205,30 @@ def get_explore_asset_spec(self, looker_structure: LookerStructureData) -> Asset breaking_version="1.10", additional_warn_text="Use `DagsterLookerApiTranslator.get_asset_spec().key` instead", ) - def get_dashboard_asset_key(self, looker_structure: LookerStructureData) -> AssetKey: + def get_dashboard_asset_key( + self, looker_structure: LookerApiTranslatorStructureData + ) -> AssetKey: return self.get_asset_spec(looker_structure).key - def get_dashboard_asset_spec(self, looker_structure: LookerStructureData) -> AssetSpec: + def get_dashboard_asset_spec( + self, looker_structure: LookerApiTranslatorStructureData + ) -> AssetSpec: looker_dashboard = check.inst(looker_structure.data, Dashboard) user = None - if self.instance_data and looker_dashboard.user_id: - user = self.instance_data.users_by_id.get(looker_dashboard.user_id) + if looker_structure.instance_data and looker_dashboard.user_id: + user = looker_structure.instance_data.users_by_id.get(looker_dashboard.user_id) return AssetSpec( key=AssetKey(f"{check.not_none(looker_dashboard.title)}_{looker_dashboard.id}"), deps=list( { self.get_asset_spec( - LookerStructureData( - structure_type=LookerStructureType.EXPLORE, data=dashboard_filter + LookerApiTranslatorStructureData( + structure_data=LookerStructureData( + structure_type=LookerStructureType.EXPLORE, data=dashboard_filter + ), + instance_data=looker_structure.instance_data, ) ).key for dashboard_filter in looker_dashboard.dashboard_filters or [] @@ -220,7 +247,7 @@ def get_dashboard_asset_spec(self, looker_structure: LookerStructureData) -> Ass ) @public - def get_asset_spec(self, looker_structure: LookerStructureData) -> AssetSpec: + def get_asset_spec(self, looker_structure: LookerApiTranslatorStructureData) -> AssetSpec: if looker_structure.structure_type == LookerStructureType.VIEW: return self.get_view_asset_spec(looker_structure) if looker_structure.structure_type == LookerStructureType.EXPLORE: @@ -235,5 +262,5 @@ def get_asset_spec(self, looker_structure: LookerStructureData) -> AssetSpec: additional_warn_text="Use `DagsterLookerApiTranslator.get_asset_spec().key` instead", ) @public - def get_asset_key(self, looker_structure: LookerStructureData) -> AssetKey: + def get_asset_key(self, looker_structure: LookerApiTranslatorStructureData) -> AssetKey: return self.get_asset_spec(looker_structure).key diff --git a/python_modules/libraries/dagster-looker/dagster_looker/api/resource.py b/python_modules/libraries/dagster-looker/dagster_looker/api/resource.py index 8e10e06855e0d..4f178c384a7c0 100644 --- a/python_modules/libraries/dagster-looker/dagster_looker/api/resource.py +++ b/python_modules/libraries/dagster-looker/dagster_looker/api/resource.py @@ -20,6 +20,7 @@ from dagster_looker.api.dagster_looker_api_translator import ( DagsterLookerApiTranslator, + LookerApiTranslatorStructureData, LookerInstanceData, LookerStructureData, LookerStructureType, @@ -177,7 +178,7 @@ def fetch_state(self) -> Mapping[str, Any]: def defs_from_state(self, state: Mapping[str, Any]) -> Definitions: looker_instance_data = LookerInstanceData.from_state(self.looker_resource.get_sdk(), state) - translator = self.translator_cls(looker_instance_data) + translator = self.translator_cls() return self._build_defs_from_looker_instance_data(looker_instance_data, translator) def _build_defs_from_looker_instance_data( @@ -187,20 +188,26 @@ def _build_defs_from_looker_instance_data( ) -> Definitions: explores = [ dagster_looker_translator.get_asset_spec( - LookerStructureData( - structure_type=LookerStructureType.EXPLORE, - data=lookml_explore, - base_url=self.looker_resource.base_url, - ), + LookerApiTranslatorStructureData( + structure_data=LookerStructureData( + structure_type=LookerStructureType.EXPLORE, + data=lookml_explore, + base_url=self.looker_resource.base_url, + ), + instance_data=looker_instance_data, + ) ) for lookml_explore in looker_instance_data.explores_by_id.values() ] views = [ dagster_looker_translator.get_asset_spec( - LookerStructureData( - structure_type=LookerStructureType.DASHBOARD, - data=looker_dashboard, - base_url=self.looker_resource.base_url, + LookerApiTranslatorStructureData( + structure_data=LookerStructureData( + structure_type=LookerStructureType.DASHBOARD, + data=looker_dashboard, + base_url=self.looker_resource.base_url, + ), + instance_data=looker_instance_data, ) ) for looker_dashboard in looker_instance_data.dashboards_by_id.values() diff --git a/python_modules/libraries/dagster-looker/dagster_looker_tests/api/test_build_defs.py b/python_modules/libraries/dagster-looker/dagster_looker_tests/api/test_build_defs.py index 035979d96ea84..2d34eaf5f9561 100644 --- a/python_modules/libraries/dagster-looker/dagster_looker_tests/api/test_build_defs.py +++ b/python_modules/libraries/dagster-looker/dagster_looker_tests/api/test_build_defs.py @@ -8,7 +8,7 @@ from dagster_looker.api.assets import build_looker_pdt_assets_definitions from dagster_looker.api.dagster_looker_api_translator import ( DagsterLookerApiTranslator, - LookerStructureData, + LookerApiTranslatorStructureData, RequestStartPdtBuild, ) from dagster_looker.api.resource import LookerResource, load_looker_asset_specs @@ -196,7 +196,7 @@ def test_custom_asset_specs( looker_resource: LookerResource, looker_instance_data_mocks: responses.RequestsMock ) -> None: class CustomDagsterLookerApiTranslator(DagsterLookerApiTranslator): - def get_asset_spec(self, looker_structure: LookerStructureData) -> AssetSpec: + def get_asset_spec(self, looker_structure: LookerApiTranslatorStructureData) -> AssetSpec: default_spec = super().get_asset_spec(looker_structure) return default_spec.replace_attributes( key=default_spec.key.with_prefix("my_prefix"),