Add isExecutable to graphql layer (#16578)

## Summary & Motivation We add isExecutable to the graphql layer by using the metadata we stashed along the way. This is the way that dagit will know that it cannot execute this asset. ## How I Tested These Changes BK. Manually loaded unexecutable asset in dagit and viewed. ![Screenshot 2023-09-18 at 5 28 09 AM](https://github.com/dagster-io/dagster/assets/28738937/af146d3e-8813-40b6-88e0-0effa22d50a6) --------- Co-authored-by: Nicholas Schrock <[email protected]>
dagster-io · Sep 18, 2023 · df1105c · df1105c · github-actions · Sep 18, 2023
1 parent eb86610
commit df1105c
Show file tree

Hide file tree

Showing 10 changed files with 6,875 additions and 5,820 deletions.
diff --git a/js_modules/dagster-ui/packages/ui-core/src/graphql/schema.graphql b/js_modules/dagster-ui/packages/ui-core/src/graphql/schema.graphql
diff --git a/js_modules/dagster-ui/packages/ui-core/src/graphql/types.ts b/js_modules/dagster-ui/packages/ui-core/src/graphql/types.ts
diff --git a/python_modules/dagster-graphql/dagster_graphql/schema/asset_graph.py b/python_modules/dagster-graphql/dagster_graphql/schema/asset_graph.py
@@ -12,6 +12,7 @@
     StaleStatus,
 )
 from dagster._core.definitions.external_asset_graph import ExternalAssetGraph
+from dagster._core.definitions.metadata import TextMetadataValue
 from dagster._core.definitions.partition import CachingDynamicPartitionsLoader, PartitionsDefinition
 from dagster._core.errors import DagsterInvariantViolationError
 from dagster._core.event_api import EventRecordsFilter
@@ -223,6 +224,7 @@ class GrapheneAssetNode(graphene.ObjectType):
     graphName = graphene.String()
     groupName = graphene.String()
     id = graphene.NonNull(graphene.ID)
+    isExecutable = graphene.NonNull(graphene.Boolean)
     isObservable = graphene.NonNull(graphene.Boolean)
     isPartitioned = graphene.NonNull(graphene.Boolean)
     isSource = graphene.NonNull(graphene.Boolean)
@@ -821,6 +823,22 @@ def resolve_isPartitioned(self, _graphene_info: ResolveInfo) -> bool:
     def resolve_isObservable(self, _graphene_info: ResolveInfo) -> bool:
         return self._external_asset_node.is_observable
 
+    def resolve_isExecutable(self, _graphene_info: ResolveInfo) -> bool:
+        from dagster._core.definitions.asset_spec import (
+            SYSTEM_METADATA_KEY_ASSET_VARIETAL,
+            AssetVarietal,
+        )
+
+        metadata_value = self._external_asset_node.metadata.get(SYSTEM_METADATA_KEY_ASSET_VARIETAL)
+        if not metadata_value:
+            varietal_text = None
+        else:
+            check.inst(metadata_value, TextMetadataValue)  # for guaranteed runtime error
+            assert isinstance(metadata_value, TextMetadataValue)  # for type checker
+            varietal_text = metadata_value.value
+
+        return AssetVarietal.is_executable(varietal_text)
+
     def resolve_latestMaterializationByPartition(
         self,
         graphene_info: ResolveInfo,