run materializations (#7394)

dagster-io · Apr 12, 2022 · 4f77c97 · 4f77c97
1 parent 96a6762
commit 4f77c97
Show file tree

Hide file tree

Showing 4 changed files with 74 additions and 0 deletions.
diff --git a/js_modules/dagit/packages/core/src/graphql/schema.graphql b/js_modules/dagit/packages/core/src/graphql/schema.graphql
diff --git a/python_modules/dagster-graphql/dagster_graphql/schema/pipelines/pipeline.py b/python_modules/dagster-graphql/dagster_graphql/schema/pipelines/pipeline.py
@@ -236,6 +236,7 @@ class GrapheneRun(graphene.ObjectType):
     rootRunId = graphene.Field(graphene.String)
     parentRunId = graphene.Field(graphene.String)
     canTerminate = graphene.NonNull(graphene.Boolean)
+    assetMaterializations = non_null_list(GrapheneMaterializationEvent)
     assets = non_null_list(GrapheneAsset)
     events = graphene.Field(
         non_null_list(GrapheneDagsterRunEvent),
@@ -357,6 +358,15 @@ def resolve_canTerminate(self, graphene_info):
     def resolve_assets(self, graphene_info):
         return get_assets_for_run_id(graphene_info, self.run_id)
 
+    def resolve_assetMaterializations(self, graphene_info):
+        # convenience field added for users querying directly via GraphQL
+        return [
+            GrapheneMaterializationEvent(event=event)
+            for event in graphene_info.context.instance.all_logs(
+                self.run_id, of_type=DagsterEventType.ASSET_MATERIALIZATION
+            )
+        ]
+
     def resolve_events(self, graphene_info, after=-1):
         events = graphene_info.context.instance.logs_after(self.run_id, cursor=after)
         events = [

diff --git a/python_modules/dagster-graphql/dagster_graphql_tests/graphql/snapshots/snap_test_assets.py b/python_modules/dagster-graphql/dagster_graphql_tests/graphql/snapshots/snap_test_assets.py
@@ -817,6 +817,42 @@
     }
 }
 
+snapshots['TestAssetAwareEventLog.test_get_run_materialization[sqlite_with_default_run_launcher_deployed_grpc_env] 1'] = {
+    'runsOrError': {
+        'results': [
+            {
+                'assetMaterializations': [
+                    {
+                        'assetKey': {
+                            'path': [
+                                'a'
+                            ]
+                        }
+                    }
+                ]
+            }
+        ]
+    }
+}
+
+snapshots['TestAssetAwareEventLog.test_get_run_materialization[sqlite_with_default_run_launcher_managed_grpc_env] 1'] = {
+    'runsOrError': {
+        'results': [
+            {
+                'assetMaterializations': [
+                    {
+                        'assetKey': {
+                            'path': [
+                                'a'
+                            ]
+                        }
+                    }
+                ]
+            }
+        ]
+    }
+}
+
 snapshots['TestAssetAwareEventLog.test_op_assets[asset_aware_instance_in_process_env] 1'] = {
     'repositoryOrError': {
         'usedSolid': {

diff --git a/python_modules/dagster-graphql/dagster_graphql_tests/graphql/test_assets.py b/python_modules/dagster-graphql/dagster_graphql_tests/graphql/test_assets.py
@@ -259,6 +259,23 @@
 """
 
 
+GET_RUN_MATERIALIZATIONS = """
+    query RunAssetsQuery {
+        runsOrError {
+            ... on Runs {
+                results {
+                    assetMaterializations {
+                        assetKey {
+                            path
+                        }
+                    }
+                }
+            }
+        }
+    }
+"""
+
+
 def _create_run(graphql_context, pipeline_name, mode="default", step_keys=None):
     selector = infer_pipeline_selector(graphql_context, pipeline_name)
     result = execute_dagster_graphql(
@@ -769,6 +786,16 @@ def get_response_by_step(response):
         assert result["asset_2"]["count"] == 6
         assert result["asset_2"]["sinceLatestMaterialization"] == False
 
+    def test_get_run_materialization(self, graphql_context, snapshot):
+        _create_run(graphql_context, "single_asset_pipeline")
+        result = execute_dagster_graphql(graphql_context, GET_RUN_MATERIALIZATIONS)
+        assert result.data
+        assert result.data["runsOrError"]
+        assert result.data["runsOrError"]["results"]
+        assert len(result.data["runsOrError"]["results"]) == 1
+        assert len(result.data["runsOrError"]["results"][0]["assetMaterializations"]) == 1
+        snapshot.assert_match(result.data)
+
 
 class TestPersistentInstanceAssetInProgress(ExecutingGraphQLContextTestMatrix):
     def test_asset_in_progress(self, graphql_context):