defend against non-existent asset_defs

dagster-io · Sep 19, 2023 · 0041a2a · 0041a2a
1 parent e526129
commit 0041a2a
Showing 1 changed file with 3 additions and 2 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
@@ -779,8 +779,9 @@ def _gen_fn():
         yield DagsterEvent.asset_materialization(step_context, materialization)
 
     asset_key, partitions = _asset_key_and_partitions_for_output(output_context)
-    if asset_key:
-        assets_def = step_context.job_def.asset_layer.assets_def_for_asset(asset_key)
+    asset_layer = step_context.job_def.asset_layer
+    if asset_key and asset_layer.has_assets_def_for_asset(asset_key):
+        assets_def = asset_layer.assets_def_for_asset(asset_key)
         execution_type = assets_def.asset_execution_type_for_asset(asset_key)
         check.invariant(
             execution_type != AssetExecutionType.UNEXECUTABLE,