feat(looker): convert looker assets to external assets

python_modules/libraries/dagster-looker/dagster_looker/asset_decorator.py

@@ -3,6 +3,7 @@
 
 from dagster import AssetsDefinition, multi_asset
 from dagster._annotations import experimental
+from dagster._core.errors import DagsterInvariantViolationError
 
 from .asset_utils import (
     build_looker_dashboard_specs,
@@ -35,11 +36,22 @@ def my_looker_project_assets(): ...
     """
     dagster_looker_translator = dagster_looker_translator or DagsterLookerTranslator()
 
-    return multi_asset(
-        compute_kind="looker",
-        specs=[
-            *build_looker_dashboard_specs(project_dir, dagster_looker_translator),
-            *build_looker_explore_specs(project_dir, dagster_looker_translator),
-            *build_looker_view_specs(project_dir, dagster_looker_translator),
-        ],
-    )
+    def looker_assets_as_external_assets(fn: Callable[..., Any]) -> AssetsDefinition:
+        @multi_asset(
+            name=fn.__name__,
+            compute_kind="looker",
+            specs=[
+                *build_looker_dashboard_specs(project_dir, dagster_looker_translator),
+                *build_looker_explore_specs(project_dir, dagster_looker_translator),
+                *build_looker_view_specs(project_dir, dagster_looker_translator),
+            ],
+        )
+        def _fn():
+            raise DagsterInvariantViolationError(
+                "You have attempted to execute an unexecutable asset computed from"
+                f" {fn.__name__}."
+            )
+
+        return _fn
+
+    return looker_assets_as_external_assets

python_modules/libraries/dagster-looker/dagster_looker/asset_utils.py

@@ -4,6 +4,10 @@
 import lkml
 import yaml
 from dagster import AssetSpec
+from dagster._core.definitions.asset_spec import (
+    SYSTEM_METADATA_KEY_ASSET_EXECUTION_TYPE,
+    AssetExecutionType,
+)
 
 from .dagster_looker_translator import DagsterLookerTranslator
 
@@ -24,7 +28,10 @@ def build_looker_dashboard_specs(
                     key=dagster_looker_translator.get_asset_key(lookml_dashboard),
                     deps=dagster_looker_translator.get_deps(lookml_dashboard),
                     description=dagster_looker_translator.get_description(lookml_dashboard),
-                    metadata=dagster_looker_translator.get_metadata(lookml_dashboard),
+                    metadata={
+                        **(dagster_looker_translator.get_metadata(lookml_dashboard) or {}),
+                        SYSTEM_METADATA_KEY_ASSET_EXECUTION_TYPE: AssetExecutionType.UNEXECUTABLE.value,
+                    },
                     group_name=dagster_looker_translator.get_group_name(lookml_dashboard),
                     owners=dagster_looker_translator.get_owners(lookml_dashboard),
                     tags=dagster_looker_translator.get_tags(lookml_dashboard),
@@ -50,7 +57,10 @@ def build_looker_explore_specs(
                     key=dagster_looker_translator.get_asset_key(lookml_explore),
                     deps=dagster_looker_translator.get_deps(lookml_explore),
                     description=dagster_looker_translator.get_description(lookml_explore),
-                    metadata=dagster_looker_translator.get_metadata(lookml_explore),
+                    metadata={
+                        **(dagster_looker_translator.get_metadata(lookml_explore) or {}),
+                        SYSTEM_METADATA_KEY_ASSET_EXECUTION_TYPE: AssetExecutionType.UNEXECUTABLE.value,
+                    },
                     group_name=dagster_looker_translator.get_group_name(lookml_explore),
                     owners=dagster_looker_translator.get_owners(lookml_explore),
                     tags=dagster_looker_translator.get_tags(lookml_explore),
@@ -76,7 +86,10 @@ def build_looker_view_specs(
                     key=dagster_looker_translator.get_asset_key(lookml_view),
                     deps=dagster_looker_translator.get_deps(lookml_view),
                     description=dagster_looker_translator.get_description(lookml_view),
-                    metadata=dagster_looker_translator.get_metadata(lookml_view),
+                    metadata={
+                        **(dagster_looker_translator.get_metadata(lookml_view) or {}),
+                        SYSTEM_METADATA_KEY_ASSET_EXECUTION_TYPE: AssetExecutionType.UNEXECUTABLE.value,
+                    },
                     group_name=dagster_looker_translator.get_group_name(lookml_view),
                     owners=dagster_looker_translator.get_owners(lookml_view),
                     tags=dagster_looker_translator.get_tags(lookml_view),

python_modules/libraries/dagster-looker/dagster_looker_tests/test_asset_decorator.py

@@ -2,7 +2,11 @@
 from typing import Any, Mapping, Optional, Sequence, Tuple
 
 import pytest
-from dagster import AssetKey
+from dagster import AssetKey, AssetSelection, DagsterInvalidDefinitionError, materialize
+from dagster._core.definitions.asset_spec import (
+    SYSTEM_METADATA_KEY_ASSET_EXECUTION_TYPE,
+    AssetExecutionType,
+)
 from dagster_looker.asset_decorator import looker_assets
 from dagster_looker.dagster_looker_translator import DagsterLookerTranslator, LookMLStructureType
 
@@ -350,7 +354,10 @@ def get_metadata(self, _) -> Optional[Mapping[str, Any]]:
     def my_looker_assets(): ...
 
     for metadata in my_looker_assets.metadata_by_key.values():
-        assert metadata == expected_metadata
+        assert metadata == {
+            **expected_metadata,
+            SYSTEM_METADATA_KEY_ASSET_EXECUTION_TYPE: AssetExecutionType.UNEXECUTABLE.value,
+        }
 
 
 def test_with_group_replacements() -> None:
@@ -402,3 +409,14 @@ def my_looker_assets(): ...
 
     for tags in my_looker_assets.tags_by_key.values():
         assert tags == expected_tags
+
+
+def test_unexecutable() -> None:
+    @looker_assets(project_dir=test_retail_demo_path)
+    def my_looker_assets(): ...
+
+    with pytest.raises(DagsterInvalidDefinitionError):
+        materialize(
+            [my_looker_assets],
+            selection=AssetSelection.assets(my_looker_assets),
+        )