[4/n] [RFC] add launch multiple runs backend functionality

python_modules/dagster-graphql/dagster_graphql/client/query.py

@@ -330,6 +330,66 @@
 )
 
 
+LAUNCH_MULTIPLE_RUNS_MUTATION = (
+    ERROR_FRAGMENT
+    + """
+mutation($executionParamsList: [ExecutionParams!]!) {
+  launchMultipleRuns(executionParamsList: $executionParamsList) {
+    __typename
+    launchMultipleRunsResult {
+      __typename
+      ... on InvalidStepError {
+        invalidStepKey
+      }
+      ... on InvalidOutputError {
+        stepKey
+        invalidOutputName
+      }
+      ... on LaunchRunSuccess {
+        run {
+          runId
+          pipeline {
+            name
+          }
+          tags {
+            key
+            value
+          }
+          status
+          runConfigYaml
+          mode
+          resolvedOpSelection
+        }
+      }
+      ... on ConflictingExecutionParamsError {
+        message
+      }
+      ... on PresetNotFoundError {
+        preset
+        message
+      }
+      ... on RunConfigValidationInvalid {
+        pipelineName
+        errors {
+          __typename
+          message
+          path
+          reason
+        }
+      }
+      ... on PipelineNotFoundError {
+        message
+        pipelineName
+      }
+      ... on PythonError {
+        ...errorFragment
+      }
+    }
+  }
+}
+"""
+)
+
 LAUNCH_PIPELINE_REEXECUTION_MUTATION = (
     ERROR_FRAGMENT
     + """

python_modules/dagster-graphql/dagster_graphql/schema/roots/__init__.py

@@ -10,6 +10,7 @@ def types():
         GrapheneDeletePipelineRunSuccess,
         GrapheneDeleteRunMutation,
         GrapheneLaunchBackfillMutation,
+        GrapheneLaunchMultipleRunsMutation,
         GrapheneLaunchRunMutation,
         GrapheneLaunchRunReexecutionMutation,
         GrapheneReloadRepositoryLocationMutation,
@@ -38,6 +39,7 @@ def types():
         GrapheneExecutionPlanOrError,
         GrapheneLaunchBackfillMutation,
         GrapheneLaunchRunMutation,
+        GrapheneLaunchMultipleRunsMutation,
         GrapheneLaunchRunReexecutionMutation,
         GraphenePipelineOrError,
         GrapheneReloadRepositoryLocationMutation,

python_modules/dagster-graphql/dagster_graphql/schema/roots/mutation.py

@@ -1,4 +1,4 @@
-from typing import Optional, Sequence, Union
+from typing import List, Optional, Sequence, Union
 
 import dagster._check as check
 import graphene
@@ -78,6 +78,7 @@
 )
 from dagster_graphql.schema.pipelines.pipeline import GrapheneRun
 from dagster_graphql.schema.runs import (
+    GrapheneLaunchMultipleRunsResult,
     GrapheneLaunchRunReexecutionResult,
     GrapheneLaunchRunResult,
     GrapheneLaunchRunSuccess,
@@ -316,6 +317,35 @@ def mutate(
         return create_execution_params_and_launch_pipeline_exec(graphene_info, executionParams)
 
 
+class GrapheneLaunchMultipleRunsMutation(graphene.Mutation):
+    """Launches multiple job runs."""
+
+    Output = graphene.NonNull(GrapheneLaunchMultipleRunsResult)
+
+    class Arguments:
+        executionParamsList = non_null_list(GrapheneExecutionParams)
+
+    class Meta:
+        name = "LaunchMultipleRunsMutation"
+
+    @capture_error
+    @require_permission_check(Permissions.LAUNCH_PIPELINE_EXECUTION)
+    def mutate(
+        self, graphene_info: ResolveInfo, executionParamsList: List[GrapheneExecutionParams]
+    ):
+        launch_multiple_runs_result = []
+
+        for execution_params in executionParamsList:
+            result = GrapheneLaunchRunMutation.mutate(
+                None, graphene_info, executionParams=execution_params
+            )
+            launch_multiple_runs_result.append(result)
+
+        return GrapheneLaunchMultipleRunsResult(
+            launchMultipleRunsResult=launch_multiple_runs_result
+        )
+
+
 class GrapheneLaunchBackfillMutation(graphene.Mutation):
     """Launches a set of partition backfill runs."""
 
@@ -984,6 +1014,7 @@ class Meta:
 
     launchPipelineExecution = GrapheneLaunchRunMutation.Field()
     launchRun = GrapheneLaunchRunMutation.Field()
+    launchMultipleRuns = GrapheneLaunchMultipleRunsMutation.Field()
     launchPipelineReexecution = GrapheneLaunchRunReexecutionMutation.Field()
     launchRunReexecution = GrapheneLaunchRunReexecutionMutation.Field()
     startSchedule = GrapheneStartScheduleMutation.Field()

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

@@ -11,6 +11,7 @@
 
 from dagster_graphql.implementation.fetch_runs import get_run_ids, get_runs, get_runs_count
 from dagster_graphql.implementation.utils import UserFacingGraphQLError
+from dagster_graphql.schema.backfill import pipeline_execution_error_types
 from dagster_graphql.schema.errors import (
     GrapheneInvalidPipelineRunsFilterError,
     GraphenePythonError,
@@ -73,17 +74,22 @@ class Meta:
 
 class GrapheneLaunchRunResult(graphene.Union):
     class Meta:
-        from dagster_graphql.schema.backfill import pipeline_execution_error_types
-
         types = launch_pipeline_run_result_types + pipeline_execution_error_types
 
         name = "LaunchRunResult"
 
 
-class GrapheneLaunchRunReexecutionResult(graphene.Union):
+class GrapheneLaunchMultipleRunsResult(graphene.ObjectType):
+    """Contains results from multiple pipeline launches."""
+
+    launchMultipleRunsResult = non_null_list(GrapheneLaunchRunResult)
+
     class Meta:
-        from dagster_graphql.schema.backfill import pipeline_execution_error_types
+        name = "LaunchMultipleRunsResult"
 
+
+class GrapheneLaunchRunReexecutionResult(graphene.Union):
+    class Meta:
         types = launch_pipeline_run_result_types + pipeline_execution_error_types
 
         name = "LaunchRunReexecutionResult"
@@ -213,6 +219,7 @@ def parse_run_config_input(
 
 types = [
     GrapheneLaunchRunResult,
+    GrapheneLaunchMultipleRunsResult,
     GrapheneLaunchRunReexecutionResult,
     GrapheneLaunchPipelineRunSuccess,
     GrapheneLaunchRunSuccess,