InputContext.upstream_output is missing key when source asset (#7919)

dagster-io · May 16, 2022 · fe59b68 · fe59b68
1 parent 5897837
commit fe59b68
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 0 deletions.
diff --git a/python_modules/dagster/dagster/core/asset_defs/assets_job.py b/python_modules/dagster/dagster/core/asset_defs/assets_job.py
@@ -322,6 +322,7 @@ def _op():
                 metadata=cast(Dict[str, Any], source_asset.metadata),
                 resource_config=resource_config,
                 resources=cast(NamedTuple, resources)._asdict(),
+                asset_key=source_asset_key,
             )
             input_context_with_upstream = build_input_context(
                 name=input_context.name,

diff --git a/python_modules/dagster/dagster_tests/core_tests/asset_defs_tests/test_assets_job.py b/python_modules/dagster/dagster_tests/core_tests/asset_defs_tests/test_assets_job.py
@@ -236,6 +236,8 @@ def load_input(self, context):
             assert context.resource_config["a"] == 7
             assert context.resources.subresource == 9
             assert context.upstream_output.resources.subresource == 9
+            assert context.upstream_output.asset_key == AssetKey("source1")
+            assert context.asset_key == AssetKey("source1")
             return 5
 
     @io_manager(config_schema={"a": int}, required_resource_keys={"subresource"})