[external-assets] Refactor from_graph_and_asset_node_mapping

dagster-io · Mar 20, 2024 · 05b414b · 05b414b
1 parent 7e08c05
commit 05b414b
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 10 deletions.
diff --git a/python_modules/dagster/dagster/_core/execution/plan/execute_step.py b/python_modules/dagster/dagster/_core/execution/plan/execute_step.py
@@ -912,11 +912,6 @@ def _log_materialization_or_observation_events_for_asset(
             )
             execution_type = AssetExecutionType.MATERIALIZATION
 
-        check.invariant(
-            execution_type != AssetExecutionType.UNEXECUTABLE,
-            "There should never be unexecutable assets here",
-        )
-
         check.invariant(
             execution_type in {AssetExecutionType.MATERIALIZATION, AssetExecutionType.OBSERVATION},
             f"Unexpected asset execution type {execution_type}",

diff --git a/python_modules/dagster/dagster_tests/definitions_tests/test_external_assets.py b/python_modules/dagster/dagster_tests/definitions_tests/test_external_assets.py
@@ -232,8 +232,6 @@ def an_asset(context: AssetExecutionContext, source_asset: str) -> str:
 
     result_two = job_def_with_shim.execute_in_process(
         instance=instance,
-        # currently we have to explicitly select the asset to exclude the source from execution
-        asset_selection=[AssetKey("an_asset")],
         partition_key="2021-01-03",
     )
 
@@ -284,9 +282,9 @@ def _downstream_asset(context: AssetExecutionContext) -> None:
     defs = Definitions(assets=[_upstream_def, _downstream_asset])
     assert defs
 
-    assert defs.get_implicit_global_asset_job_def().asset_layer.asset_deps[
-        AssetKey("downstream_asset")
-    ] == {AssetKey("upstream_asset")}
+    assert defs.get_asset_graph().asset_dep_graph["upstream"][AssetKey("downstream_asset")] == {
+        AssetKey("upstream_asset")
+    }
 
 
 def test_external_asset_multi_asset() -> None: