expose compute_kind tag value on gql AssetNode (#7120)

dagster-io · Mar 18, 2022 · 8aaa2e6 · 8aaa2e6
1 parent 5efccff
commit 8aaa2e6
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 3 deletions.
diff --git a/python_modules/dagster-graphql/dagster_graphql/schema/asset_graph.py b/python_modules/dagster-graphql/dagster_graphql/schema/asset_graph.py
@@ -78,6 +78,7 @@ class GrapheneAssetNode(graphene.ObjectType):
         beforeTimestampMillis=graphene.String(),
         limit=graphene.Int(),
     )
+    computeKind: graphene.String
     dependedBy = non_null_list(GrapheneAssetDependency)
     dependedByKeys = non_null_list(GrapheneAssetKey)
     dependencies = non_null_list(GrapheneAssetDependency)
@@ -198,6 +199,9 @@ def resolve_assetMaterializations(
             )
         ]
 
+    def resolve_computeKind(self, _graphene_info) -> Optional[str]:
+        return self._external_asset_node.compute_kind
+
     def resolve_dependedBy(self, graphene_info) -> List[GrapheneAssetDependency]:
         if not self._external_asset_node.depended_by:
             return []
@@ -249,7 +253,7 @@ def resolve_dependencies(self, graphene_info) -> List[GrapheneAssetDependency]:
             for dep in self._external_asset_node.dependencies
         ]
 
-    def resolve_jobNames(self, _graphene_info) -> List[graphene.String]:
+    def resolve_jobNames(self, _graphene_info) -> List[str]:
         return self._external_asset_node.job_names
 
     def resolve_jobs(self, _graphene_info) -> List[GraphenePipeline]:
@@ -324,13 +328,13 @@ def resolve_op(
         else:
             return None
 
-    def resolve_partitionDefinition(self, _graphene_info) -> Optional[graphene.String]:
+    def resolve_partitionDefinition(self, _graphene_info) -> Optional[str]:
         partitions_def_data = self._external_asset_node.partitions_def_data
         if partitions_def_data:
             return str(partitions_def_data.get_partitions_definition())
         return None
 
-    def resolve_partitionKeys(self, _graphene_info) -> List[graphene.String]:
+    def resolve_partitionKeys(self, _graphene_info) -> List[str]:
         return self.get_partition_keys()
 
     def resolve_repository(self, graphene_info) -> "GrapheneRepository":

diff --git a/python_modules/dagster/dagster/core/host_representation/external_data.py b/python_modules/dagster/dagster/core/host_representation/external_data.py
@@ -687,6 +687,7 @@ class ExternalAssetNode(
             ("asset_key", AssetKey),
             ("dependencies", Sequence[ExternalAssetDependency]),
             ("depended_by", Sequence[ExternalAssetDependedBy]),
+            ("compute_kind", Optional[str]),
             ("op_name", Optional[str]),
             ("op_description", Optional[str]),
             ("job_names", Sequence[str]),
@@ -707,6 +708,7 @@ def __new__(
         asset_key: AssetKey,
         dependencies: Sequence[ExternalAssetDependency],
         depended_by: Sequence[ExternalAssetDependedBy],
+        compute_kind: Optional[str] = None,
         op_name: Optional[str] = None,
         op_description: Optional[str] = None,
         job_names: Optional[Sequence[str]] = None,
@@ -724,6 +726,7 @@ def __new__(
             depended_by=check.opt_sequence_param(
                 depended_by, "depended_by", of_type=ExternalAssetDependedBy
             ),
+            compute_kind=check.opt_str_param(compute_kind, "compute_kind"),
             op_name=check.opt_str_param(op_name, "op_name"),
             op_description=check.opt_str_param(
                 op_description or output_description, "op_description"
@@ -893,6 +896,7 @@ def external_asset_graph_from_defs(
                 asset_key=asset_key,
                 dependencies=list(deps[asset_key].values()),
                 depended_by=list(dep_by[asset_key].values()),
+                compute_kind=node_def.tags.get("kind"),
                 op_name=node_def.name,
                 op_description=node_def.description,
                 job_names=job_names,