canExecuteIndividually on asset checks (dagster-io#16509)

To start, disable executing if the user code version is < 1.4.14 (the
version we're shipping with support for it tomorrow).

A follow up will add checks against checks+assets in the same op

python_modules/dagster-graphql/dagster_graphql/implementation/fetch_asset_checks.py

@@ -11,9 +11,11 @@
     AssetCheckExecutionResolvedStatus,
 )
 from dagster._core.storage.dagster_run import DagsterRunStatus
+from packaging import version
 
 from ..schema.asset_checks import (
     GrapheneAssetCheck,
+    GrapheneAssetCheckCanExecuteIndividually,
     GrapheneAssetCheckExecution,
     GrapheneAssetCheckNeedsMigrationError,
     GrapheneAssetChecks,
@@ -30,14 +32,26 @@ def _fetch_asset_checks(
 ) -> GrapheneAssetChecks:
     external_asset_checks = []
     for location in graphene_info.context.code_locations:
+        # check if the code location is too old to support executing asset checks individually
+        code_location_version = (location.get_dagster_library_versions() or {}).get("dagster")
+        if code_location_version and version.parse(code_location_version) < version.parse("1.4.14"):
+            can_execute_individually = (
+                GrapheneAssetCheckCanExecuteIndividually.NEEDS_USER_CODE_UPGRADE
+            )
+        else:
+            can_execute_individually = GrapheneAssetCheckCanExecuteIndividually.CAN_EXECUTE
+
         for repository in location.get_repositories().values():
             for external_check in repository.external_repository_data.external_asset_checks or []:
                 if external_check.asset_key == asset_key:
                     if not check_name or check_name == external_check.name:
-                        external_asset_checks.append(external_check)
+                        external_asset_checks.append((external_check, can_execute_individually))
 
     return GrapheneAssetChecks(
-        checks=[GrapheneAssetCheck(check) for check in external_asset_checks]
+        checks=[
+            GrapheneAssetCheck(asset_check=check, can_execute_individually=can_execute_individually)
+            for check, can_execute_individually in external_asset_checks
+        ]
     )
 
 

python_modules/dagster-graphql/dagster_graphql/schema/asset_checks.py

@@ -108,6 +108,14 @@ def __init__(
         self.timestamp = execution.create_timestamp
 
 
+class GrapheneAssetCheckCanExecuteIndividually(graphene.Enum):
+    class Meta:
+        name = "AssetCheckCanExecuteIndividually"
+
+    CAN_EXECUTE = "CAN_EXECUTE"
+    NEEDS_USER_CODE_UPGRADE = "NEEDS_USER_CODE_UPGRADE"
+
+
 class GrapheneAssetCheck(graphene.ObjectType):
     name = graphene.NonNull(graphene.String)
     assetKey = graphene.NonNull(GrapheneAssetKey)
@@ -118,12 +126,18 @@ class GrapheneAssetCheck(graphene.ObjectType):
         cursor=graphene.String(),
     )
     executionForLatestMaterialization = graphene.Field(GrapheneAssetCheckExecution)
+    canExecuteIndividually = graphene.NonNull(GrapheneAssetCheckCanExecuteIndividually)
 
     class Meta:
         name = "AssetCheck"
 
-    def __init__(self, asset_check: ExternalAssetCheck):
+    def __init__(
+        self,
+        asset_check: ExternalAssetCheck,
+        can_execute_individually: GrapheneAssetCheckCanExecuteIndividually,
+    ):
         self._asset_check = asset_check
+        self._can_execute_individually = can_execute_individually
 
     def resolve_assetKey(self, _):
         return self._asset_check.asset_key
@@ -156,6 +170,9 @@ def resolve_executionForLatestMaterialization(
             graphene_info.context.instance, self._asset_check
         )
 
+    def resolve_canExecuteIndividually(self, _) -> GrapheneAssetCheckCanExecuteIndividually:
+        return self._can_execute_individually
+
 
 class GrapheneAssetChecks(graphene.ObjectType):
     checks = non_null_list(GrapheneAssetCheck)

python_modules/dagster-graphql/dagster_graphql_tests/graphql/test_asset_checks.py

@@ -34,6 +34,7 @@
                     path
                 }
                 description
+                canExecuteIndividually
             }
         }
     }
@@ -254,6 +255,7 @@ def test_asset_check_definitions(self, graphql_context: WorkspaceRequestContext)
                             "path": ["asset_1"],
                         },
                         "description": "asset_1 check",
+                        "canExecuteIndividually": "CAN_EXECUTE",
                     }
                 ]
             }