From 7e68f710b155116477bd2c30650bb94115f79404 Mon Sep 17 00:00:00 2001 From: Maxime Armstrong Date: Fri, 27 Dec 2024 14:31:38 -0500 Subject: [PATCH 1/8] [dagster-looker] Use Looker API translator instance in spec loader and state-backed defs --- .../looker/customize-looker-assets.py | 2 +- .../dagster_looker/api/assets.py | 11 ++++---- .../dagster_looker/api/resource.py | 26 ++++++++----------- .../api/test_build_defs.py | 2 +- 4 files changed, 19 insertions(+), 22 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 981297d84a1a2..9aa706f56eb57 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 @@ -32,6 +32,6 @@ def get_asset_spec(self, looker_structure: LookerStructureData) -> dg.AssetSpec: looker_specs = load_looker_asset_specs( - looker_resource, dagster_looker_translator=CustomDagsterLookerApiTranslator + looker_resource, dagster_looker_translator=CustomDagsterLookerApiTranslator # type: ignore ) defs = dg.Definitions(assets=[*looker_specs], resources={"looker": looker_resource}) 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 299cd33ccb380..c5c173ee83fcc 100644 --- a/python_modules/libraries/dagster-looker/dagster_looker/api/assets.py +++ b/python_modules/libraries/dagster-looker/dagster_looker/api/assets.py @@ -1,4 +1,4 @@ -from typing import Sequence, Type, cast +from typing import Optional, Sequence, cast from dagster import AssetExecutionContext, AssetsDefinition, Failure, multi_asset from dagster._annotations import experimental @@ -18,7 +18,7 @@ def build_looker_pdt_assets_definitions( resource_key: str, request_start_pdt_builds: Sequence[RequestStartPdtBuild], - dagster_looker_translator: Type[DagsterLookerApiTranslator] = DagsterLookerApiTranslator, + dagster_looker_translator: Optional[DagsterLookerApiTranslator] = None, ) -> Sequence[AssetsDefinition]: """Returns the AssetsDefinitions of the executable assets for the given the list of refreshable PDTs. @@ -27,13 +27,14 @@ def build_looker_pdt_assets_definitions( request_start_pdt_builds (Optional[Sequence[RequestStartPdtBuild]]): A list of requests to start PDT builds. See https://developers.looker.com/api/explorer/4.0/types/DerivedTable/RequestStartPdtBuild?sdk=py for documentation on all available fields. - dagster_looker_translator (Optional[DagsterLookerApiTranslator]): The translator to - use to convert Looker structures into assets. Defaults to DagsterLookerApiTranslator. + dagster_looker_translator (Optional[DagsterLookerApiTranslator]): The translator to use + to convert Looker structures into :py:class:`dagster.AssetSpec`. + Defaults to :py:class:`DagsterLookerApiTranslator`. Returns: AssetsDefinition: The AssetsDefinitions of the executable assets for the given the list of refreshable PDTs. """ - translator = dagster_looker_translator() + translator = dagster_looker_translator or DagsterLookerApiTranslator() result = [] for request_start_pdt_build in request_start_pdt_builds: 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 4f178c384a7c0..fbf456c2d2785 100644 --- a/python_modules/libraries/dagster-looker/dagster_looker/api/resource.py +++ b/python_modules/libraries/dagster-looker/dagster_looker/api/resource.py @@ -1,6 +1,6 @@ from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Sequence, Tuple, Type, cast +from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Sequence, Tuple, cast from dagster import ( AssetSpec, @@ -107,20 +107,16 @@ def build_defs( from dagster_looker.api.assets import build_looker_pdt_assets_definitions resource_key = "looker" - translator_cls = ( - dagster_looker_translator.__class__ - if dagster_looker_translator - else DagsterLookerApiTranslator - ) + translator = dagster_looker_translator or DagsterLookerApiTranslator() pdts = build_looker_pdt_assets_definitions( resource_key=resource_key, request_start_pdt_builds=request_start_pdt_builds or [], - dagster_looker_translator=translator_cls, + dagster_looker_translator=translator, ) return Definitions( - assets=[*pdts, *load_looker_asset_specs(self, translator_cls, looker_filter)], + assets=[*pdts, *load_looker_asset_specs(self, translator, looker_filter)], resources={resource_key: self}, ) @@ -128,15 +124,16 @@ def build_defs( @experimental def load_looker_asset_specs( looker_resource: LookerResource, - dagster_looker_translator: Type[DagsterLookerApiTranslator] = DagsterLookerApiTranslator, + dagster_looker_translator: Optional[DagsterLookerApiTranslator] = None, looker_filter: Optional[LookerFilter] = None, ) -> Sequence[AssetSpec]: """Returns a list of AssetSpecs representing the Looker structures. Args: looker_resource (LookerResource): The Looker resource to fetch assets from. - dagster_looker_translator (Type[DagsterLookerApiTranslator]): The translator to use - to convert Looker structures into AssetSpecs. Defaults to DagsterLookerApiTranslator. + dagster_looker_translator (Optional[DagsterLookerApiTranslator]): The translator to use + to convert Looker structures into :py:class:`dagster.AssetSpec`. + Defaults to :py:class:`DagsterLookerApiTranslator`. Returns: List[AssetSpec]: The set of AssetSpecs representing the Looker structures. @@ -144,7 +141,7 @@ def load_looker_asset_specs( return check.is_list( LookerApiDefsLoader( looker_resource=looker_resource, - translator_cls=dagster_looker_translator, + translator=dagster_looker_translator or DagsterLookerApiTranslator(), looker_filter=looker_filter or LookerFilter(), ) .build_defs() @@ -165,7 +162,7 @@ def build_folder_path(folder_id_to_folder: Dict[str, "Folder"], folder_id: str) @dataclass(frozen=True) class LookerApiDefsLoader(StateBackedDefinitionsLoader[Mapping[str, Any]]): looker_resource: LookerResource - translator_cls: Type[DagsterLookerApiTranslator] + translator: DagsterLookerApiTranslator looker_filter: LookerFilter @property @@ -178,8 +175,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() - return self._build_defs_from_looker_instance_data(looker_instance_data, translator) + return self._build_defs_from_looker_instance_data(looker_instance_data, self.translator) def _build_defs_from_looker_instance_data( self, 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 2d34eaf5f9561..c99332fffc17d 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 @@ -205,7 +205,7 @@ def get_asset_spec(self, looker_structure: LookerApiTranslatorStructureData) -> all_assets = ( asset for asset in Definitions( - assets=[*load_looker_asset_specs(looker_resource, CustomDagsterLookerApiTranslator)], + assets=[*load_looker_asset_specs(looker_resource, CustomDagsterLookerApiTranslator())], ) .get_asset_graph() .assets_defs From 33b7453cdb2da8bb58eb955aca241f2581e88447 Mon Sep 17 00:00:00 2001 From: Maxime Armstrong Date: Fri, 27 Dec 2024 15:08:02 -0500 Subject: [PATCH 2/8] Lint --- docs/content/integrations/looker.mdx | 3 ++- .../integrations/looker/customize-looker-assets.py | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/docs/content/integrations/looker.mdx b/docs/content/integrations/looker.mdx index 03ebb764a45df..b901c8ad6e3d6 100644 --- a/docs/content/integrations/looker.mdx +++ b/docs/content/integrations/looker.mdx @@ -125,7 +125,8 @@ class CustomDagsterLookerApiTranslator(DagsterLookerApiTranslator): looker_specs = load_looker_asset_specs( - looker_resource, dagster_looker_translator=CustomDagsterLookerApiTranslator + looker_resource, + dagster_looker_translator=CustomDagsterLookerApiTranslator, ) defs = dg.Definitions(assets=[*looker_specs], resources={"looker": looker_resource}) ``` 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 9aa706f56eb57..e10a286d8d6ba 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 @@ -32,6 +32,7 @@ def get_asset_spec(self, looker_structure: LookerStructureData) -> dg.AssetSpec: looker_specs = load_looker_asset_specs( - looker_resource, dagster_looker_translator=CustomDagsterLookerApiTranslator # type: ignore + looker_resource, + dagster_looker_translator=CustomDagsterLookerApiTranslator, # type: ignore ) defs = dg.Definitions(assets=[*looker_specs], resources={"looker": looker_resource}) From b3415f810d85481df80884a564db4c6cf4fe129f Mon Sep 17 00:00:00 2001 From: Maxime Armstrong Date: Thu, 2 Jan 2025 12:00:59 -0500 Subject: [PATCH 3/8] Add deprecation warning --- .../dagster_looker/api/resource.py | 33 ++++++++++++++++--- .../api/test_build_defs.py | 30 +++++++++++++++++ 2 files changed, 59 insertions(+), 4 deletions(-) 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 fbf456c2d2785..0077720dd14b0 100644 --- a/python_modules/libraries/dagster-looker/dagster_looker/api/resource.py +++ b/python_modules/libraries/dagster-looker/dagster_looker/api/resource.py @@ -1,6 +1,18 @@ from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Sequence, Tuple, cast +from typing import ( + TYPE_CHECKING, + Any, + Dict, + List, + Mapping, + Optional, + Sequence, + Tuple, + Type, + Union, + cast, +) from dagster import ( AssetSpec, @@ -13,6 +25,7 @@ from dagster._record import record from dagster._utils.cached_method import cached_method from dagster._utils.log import get_dagster_logger +from dagster._utils.warnings import deprecation_warning from looker_sdk import init40 from looker_sdk.rtl.api_settings import ApiSettings, SettingsConfig from looker_sdk.sdk.api40.methods import Looker40SDK @@ -124,20 +137,32 @@ def build_defs( @experimental def load_looker_asset_specs( looker_resource: LookerResource, - dagster_looker_translator: Optional[DagsterLookerApiTranslator] = None, + dagster_looker_translator: Optional[ + Union[DagsterLookerApiTranslator, Type[DagsterLookerApiTranslator]] + ] = None, looker_filter: Optional[LookerFilter] = None, ) -> Sequence[AssetSpec]: """Returns a list of AssetSpecs representing the Looker structures. Args: looker_resource (LookerResource): The Looker resource to fetch assets from. - dagster_looker_translator (Optional[DagsterLookerApiTranslator]): The translator to use - to convert Looker structures into :py:class:`dagster.AssetSpec`. + dagster_looker_translator (Optional[Union[DagsterLookerApiTranslator, Type[DagsterLookerApiTranslator]]]): + The translator to use to convert Looker structures into :py:class:`dagster.AssetSpec`. Defaults to :py:class:`DagsterLookerApiTranslator`. Returns: List[AssetSpec]: The set of AssetSpecs representing the Looker structures. """ + if isinstance(dagster_looker_translator, type): + deprecation_warning( + subject="Support of `dagster_looker_translator` as a Type[DagsterLookerApiTranslator]", + breaking_version="1.10", + additional_warn_text=( + "Pass an instance of DagsterLookerApiTranslator or subclass to `dagster_looker_translator` instead." + ), + ) + dagster_looker_translator = dagster_looker_translator() + return check.is_list( LookerApiDefsLoader( looker_resource=looker_resource, 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 c99332fffc17d..aea8816cfde39 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 @@ -219,3 +219,33 @@ def get_asset_spec(self, looker_structure: LookerApiTranslatorStructureData) -> assert all(key.path[0] == "my_prefix" for key in asset.keys) for deps in asset.asset_deps.values(): assert all(key.path[0] == "my_prefix" for key in deps), str(deps) + + +@responses.activate +def test_custom_asset_specs_legacy( + looker_resource: LookerResource, looker_instance_data_mocks: responses.RequestsMock +) -> None: + class CustomDagsterLookerApiTranslator(DagsterLookerApiTranslator): + 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"), + ).merge_attributes(metadata={"custom": "metadata"}) + + all_assets = ( + asset + for asset in Definitions( + assets=[*load_looker_asset_specs(looker_resource, CustomDagsterLookerApiTranslator)], + ) + .get_asset_graph() + .assets_defs + if not asset.is_auto_created_stub + ) + + for asset in all_assets: + for metadata in asset.metadata_by_key.values(): + assert "custom" in metadata + assert metadata["custom"] == "metadata" + assert all(key.path[0] == "my_prefix" for key in asset.keys) + for deps in asset.asset_deps.values(): + assert all(key.path[0] == "my_prefix" for key in deps), str(deps) From 82bb0802ab332a4e73f7920b99b6923a192c06e4 Mon Sep 17 00:00:00 2001 From: Maxime Armstrong Date: Thu, 2 Jan 2025 12:05:51 -0500 Subject: [PATCH 4/8] Add deprecation warning for asset factory --- .../dagster_looker/api/assets.py | 21 +++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) 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 c5c173ee83fcc..5ef551b08d5a4 100644 --- a/python_modules/libraries/dagster-looker/dagster_looker/api/assets.py +++ b/python_modules/libraries/dagster-looker/dagster_looker/api/assets.py @@ -1,7 +1,8 @@ -from typing import Optional, Sequence, cast +from typing import Optional, Sequence, cast, Union, Type from dagster import AssetExecutionContext, AssetsDefinition, Failure, multi_asset from dagster._annotations import experimental +from dagster._utils.warnings import deprecation_warning from dagster_looker.api.dagster_looker_api_translator import ( DagsterLookerApiTranslator, @@ -18,7 +19,9 @@ def build_looker_pdt_assets_definitions( resource_key: str, request_start_pdt_builds: Sequence[RequestStartPdtBuild], - dagster_looker_translator: Optional[DagsterLookerApiTranslator] = None, + dagster_looker_translator: Optional[ + Union[DagsterLookerApiTranslator, Type[DagsterLookerApiTranslator]] + ] = None, ) -> Sequence[AssetsDefinition]: """Returns the AssetsDefinitions of the executable assets for the given the list of refreshable PDTs. @@ -27,13 +30,23 @@ def build_looker_pdt_assets_definitions( request_start_pdt_builds (Optional[Sequence[RequestStartPdtBuild]]): A list of requests to start PDT builds. See https://developers.looker.com/api/explorer/4.0/types/DerivedTable/RequestStartPdtBuild?sdk=py for documentation on all available fields. - dagster_looker_translator (Optional[DagsterLookerApiTranslator]): The translator to use - to convert Looker structures into :py:class:`dagster.AssetSpec`. + dagster_looker_translator (Optional[Union[DagsterLookerApiTranslator, Type[DagsterLookerApiTranslator]]]): + The translator to use to convert Looker structures into :py:class:`dagster.AssetSpec`. Defaults to :py:class:`DagsterLookerApiTranslator`. Returns: AssetsDefinition: The AssetsDefinitions of the executable assets for the given the list of refreshable PDTs. """ + if isinstance(dagster_looker_translator, type): + deprecation_warning( + subject="Support of `dagster_looker_translator` as a Type[DagsterLookerApiTranslator]", + breaking_version="1.10", + additional_warn_text=( + "Pass an instance of DagsterLookerApiTranslator or subclass to `dagster_looker_translator` instead." + ), + ) + dagster_looker_translator = dagster_looker_translator() + translator = dagster_looker_translator or DagsterLookerApiTranslator() result = [] for request_start_pdt_build in request_start_pdt_builds: From f828454abf1dfdb9d4fd4e1d1993c21bd0fd0600 Mon Sep 17 00:00:00 2001 From: Maxime Armstrong Date: Thu, 2 Jan 2025 12:07:34 -0500 Subject: [PATCH 5/8] Lint --- .../libraries/dagster-looker/dagster_looker/api/assets.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 5ef551b08d5a4..ecadebdf33fde 100644 --- a/python_modules/libraries/dagster-looker/dagster_looker/api/assets.py +++ b/python_modules/libraries/dagster-looker/dagster_looker/api/assets.py @@ -1,4 +1,4 @@ -from typing import Optional, Sequence, cast, Union, Type +from typing import Optional, Sequence, Type, Union, cast from dagster import AssetExecutionContext, AssetsDefinition, Failure, multi_asset from dagster._annotations import experimental From 048ed44ee4671ce476e458590582ac5129eab52a Mon Sep 17 00:00:00 2001 From: Maxime Armstrong Date: Thu, 2 Jan 2025 12:08:22 -0500 Subject: [PATCH 6/8] Add comment --- .../dagster-looker/dagster_looker_tests/api/test_build_defs.py | 1 + 1 file changed, 1 insertion(+) 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 aea8816cfde39..b4252187540b6 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 @@ -232,6 +232,7 @@ def get_asset_spec(self, looker_structure: LookerApiTranslatorStructureData) -> key=default_spec.key.with_prefix("my_prefix"), ).merge_attributes(metadata={"custom": "metadata"}) + # Pass the translator type all_assets = ( asset for asset in Definitions( From 8ebd00bb3c4b1bdf00886bebe1322ff65d55db1c Mon Sep 17 00:00:00 2001 From: Maxime Armstrong Date: Thu, 2 Jan 2025 13:07:09 -0500 Subject: [PATCH 7/8] Fix Pyright --- .../integrations/looker/customize-looker-assets.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 e10a286d8d6ba..95d8e90846f81 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 @@ -33,6 +33,6 @@ def get_asset_spec(self, looker_structure: LookerStructureData) -> dg.AssetSpec: looker_specs = load_looker_asset_specs( looker_resource, - dagster_looker_translator=CustomDagsterLookerApiTranslator, # type: ignore + dagster_looker_translator=CustomDagsterLookerApiTranslator, ) defs = dg.Definitions(assets=[*looker_specs], resources={"looker": looker_resource}) From fddc7780ad8279bd0b6e19f1de78aff9223086ee Mon Sep 17 00:00:00 2001 From: Maxime Armstrong Date: Thu, 2 Jan 2025 14:57:19 -0500 Subject: [PATCH 8/8] Update post review --- .../api/test_build_defs.py | 22 ++++++++++++------- 1 file changed, 14 insertions(+), 8 deletions(-) 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 b4252187540b6..ede341ba2d686 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 @@ -233,15 +233,21 @@ def get_asset_spec(self, looker_structure: LookerApiTranslatorStructureData) -> ).merge_attributes(metadata={"custom": "metadata"}) # Pass the translator type - all_assets = ( - asset - for asset in Definitions( - assets=[*load_looker_asset_specs(looker_resource, CustomDagsterLookerApiTranslator)], + with pytest.warns( + DeprecationWarning, + match=r"Support of `dagster_looker_translator` as a Type\[DagsterLookerApiTranslator\]", + ): + all_assets = ( + asset + for asset in Definitions( + assets=[ + *load_looker_asset_specs(looker_resource, CustomDagsterLookerApiTranslator) + ], + ) + .get_asset_graph() + .assets_defs + if not asset.is_auto_created_stub ) - .get_asset_graph() - .assets_defs - if not asset.is_auto_created_stub - ) for asset in all_assets: for metadata in asset.metadata_by_key.values():