update for new field on state obj

Author: jamiedemaria

js_modules/dagster-ui/packages/ui-core/src/graphql/schema.graphql

@@ -1,5 +1,4 @@
 import asyncio
-from typing import Optional
 
 import graphene
 from dagster._core.definitions.asset_health.asset_check_health import (
@@ -23,7 +22,6 @@
     AssetHealthMaterializationMetadata,
     get_materialization_status_and_metadata,
 )
-from dagster._core.storage.event_log.base import AssetRecord
 
 from dagster_graphql.schema.entity_key import GrapheneAssetKey
 from dagster_graphql.schema.util import ResolveInfo
@@ -94,37 +92,11 @@ class GrapheneAssetHealthMaterializationDegradedPartitionedMeta(graphene.ObjectT
     numMissingPartitions = graphene.NonNull(graphene.Int)
     totalNumPartitions = graphene.NonNull(graphene.Int)
     latestRunId = graphene.String()
-    failedRunId = graphene.String()
+    latestFailedRunId = graphene.String()
 
     class Meta:
         name = "AssetHealthMaterializationDegradedPartitionedMeta"
 
-    def __init__(
-        self,
-        asset_key: GrapheneAssetKey,
-        num_failed_partitions: int,
-        num_missing_partitions: int,
-        total_num_partitions: int,
-        latest_run_id: str,
-    ):
-        super().__init__()
-        self._asset_key = asset_key
-        self.numFailedPartitions = num_failed_partitions
-        self.numMissingPartitions = num_missing_partitions
-        self.totalNumPartitions = total_num_partitions
-        self.latestRunId = latest_run_id
-
-    async def resolve_failedRunId(self, graphene_info: ResolveInfo) -> Optional[str]:
-        # The latest failed run for a degraded partitioned asset may not be the most recent run that
-        # attempted to materialize the asset. The failed partitions could be in a historical run
-        asset_record = await AssetRecord.gen(graphene_info.context, self._asset_key)
-        if (
-            asset_record is not None
-            and asset_record.asset_entry.last_failed_to_materialize_record is not None
-        ):
-            return asset_record.asset_entry.last_failed_to_materialize_record.run_id
-        return None
-
 
 class GrapheneAssetHealthMaterializationHealthyPartitionedMeta(graphene.ObjectType):
     numMissingPartitions = graphene.NonNull(graphene.Int)
@@ -158,7 +130,7 @@ def from_metadata_class(
     ) -> "GrapheneAssetHealthMaterializationMeta":
         if isinstance(metadata, AssetHealthMaterializationDegradedNotPartitionedMeta):
             return GrapheneAssetHealthMaterializationDegradedNotPartitionedMeta(
-                failedRunId=metadata.failed_run_id
+                failedRunId=metadata.latest_failed_to_materialize_run_id,
             )
         elif isinstance(metadata, AssetHealthMaterializationHealthyPartitionedMeta):
             return GrapheneAssetHealthMaterializationHealthyPartitionedMeta(
@@ -168,11 +140,11 @@ def from_metadata_class(
             )
         elif isinstance(metadata, AssetHealthMaterializationDegradedPartitionedMeta):
             return GrapheneAssetHealthMaterializationDegradedPartitionedMeta(
-                asset_key=asset_key,
-                num_failed_partitions=metadata.num_failed_partitions,
-                num_missing_partitions=metadata.num_missing_partitions,
-                total_num_partitions=metadata.total_num_partitions,
-                latest_run_id=metadata.latest_run_id,
+                numFailedPartitions=metadata.num_failed_partitions,
+                numMissingPartitions=metadata.num_missing_partitions,
+                totalNumPartitions=metadata.total_num_partitions,
+                latestRunId=metadata.latest_run_id,
+                latestFailedRunId=metadata.latest_failed_to_materialize_run_id,
             )
         else:
             raise ValueError(f"Unknown metadata class: {type(metadata)}")

js_modules/dagster-ui/packages/ui-core/src/graphql/types.ts

@@ -332,6 +332,7 @@ class AssetHealthMaterializationDegradedPartitionedMeta:
     num_missing_partitions: int
     total_num_partitions: int
     latest_run_id: Optional[str]
+    latest_failed_to_materialize_run_id: Optional[str]
 
 
 @whitelist_for_serdes
@@ -340,12 +341,13 @@ class AssetHealthMaterializationHealthyPartitionedMeta:
     num_missing_partitions: int
     total_num_partitions: int
     latest_run_id: Optional[str]
+    latest_failed_to_materialize_run_id: Optional[str]
 
 
 @whitelist_for_serdes
 @record.record
 class AssetHealthMaterializationDegradedNotPartitionedMeta:
-    failed_run_id: Optional[str]
+    latest_failed_to_materialize_run_id: Optional[str]
 
 
 AssetHealthMaterializationMetadata = Union[
@@ -442,6 +444,7 @@ async def get_materialization_status_and_metadata(
                 num_missing_partitions=num_missing,
                 total_num_partitions=total_num_partitions,
                 latest_run_id=asset_materialization_health_state.latest_terminal_run_id,
+                latest_failed_to_materialize_run_id=asset_materialization_health_state.latest_failed_to_materialize_run_id,
             )
         else:
             # captures the case when asset is not partitioned, or the asset is partitioned and all partitions are materialized
@@ -463,10 +466,11 @@ async def get_materialization_status_and_metadata(
                 num_missing_partitions=num_missing,
                 total_num_partitions=total_num_partitions,
                 latest_run_id=asset_materialization_health_state.latest_terminal_run_id,
+                latest_failed_to_materialize_run_id=asset_materialization_health_state.latest_failed_to_materialize_run_id,
             )
         else:
             meta = AssetHealthMaterializationDegradedNotPartitionedMeta(
-                failed_run_id=asset_materialization_health_state.latest_terminal_run_id,
+                latest_failed_to_materialize_run_id=asset_materialization_health_state.latest_failed_to_materialize_run_id,
             )
         return AssetHealthStatus.DEGRADED, meta
     elif asset_materialization_health_state.health_status == AssetHealthStatus.UNKNOWN: