Fix error when dagit passes in a non-dict string to run config valiat…

…ion (#8020)

Summary:
Before, this was raising an exception due to being unable to parse the JSON. Pass it through un-JSON-parsed instead so that it will get processed as a useful validation error instead of an uncaught graphql exception.

Test Plan: New BK test that was failing before, simulate in Launchpad and see a useful error

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

@@ -109,7 +109,9 @@ def create_execution_params(graphene_info, graphql_execution_params):
 def execution_params_from_graphql(graphql_execution_params):
     return ExecutionParams(
         selector=pipeline_selector_from_graphql(graphql_execution_params.get("selector")),
-        run_config=parse_run_config_input(graphql_execution_params.get("runConfigData") or {}),
+        run_config=parse_run_config_input(
+            graphql_execution_params.get("runConfigData") or {}, raise_on_error=True
+        ),
         mode=graphql_execution_params.get("mode"),
         execution_metadata=create_execution_metadata(
             graphql_execution_params.get("executionMetadata")

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

@@ -419,15 +419,15 @@ def resolve_isPipelineConfigValid(self, graphene_info, pipeline, **kwargs):
         return validate_pipeline_config(
             graphene_info,
             pipeline_selector_from_graphql(pipeline),
-            parse_run_config_input(kwargs.get("runConfigData", {})),
+            parse_run_config_input(kwargs.get("runConfigData", {}), raise_on_error=False),
             kwargs.get("mode"),
         )
 
     def resolve_executionPlanOrError(self, graphene_info, pipeline, **kwargs):
         return get_execution_plan(
             graphene_info,
             pipeline_selector_from_graphql(pipeline),
-            parse_run_config_input(kwargs.get("runConfigData", {})),
+            parse_run_config_input(kwargs.get("runConfigData", {}), raise_on_error=True),
             kwargs.get("mode"),
         )
 

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

@@ -76,7 +76,7 @@ def resolve_isRunConfigValid(self, graphene_info, **kwargs):
             graphene_info,
             self._represented_pipeline,
             self._mode,
-            parse_run_config_input(kwargs.get("runConfigData", {})),
+            parse_run_config_input(kwargs.get("runConfigData", {}), raise_on_error=False),
         )
 
 

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

@@ -144,8 +144,16 @@ class Meta:
         name = "RunConfigData"
 
 
-def parse_run_config_input(run_config):
-    return json.loads(run_config) if (run_config and isinstance(run_config, str)) else run_config
+def parse_run_config_input(run_config, raise_on_error: bool):
+    if run_config and isinstance(run_config, str):
+        try:
+            return json.loads(run_config)
+        except json.JSONDecodeError:
+            if raise_on_error:
+                raise
+            # Pass the config through as a string so that it will return a useful validation error
+            return run_config
+    return run_config
 
 
 types = [

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

@@ -168,6 +168,19 @@ def test_basic_valid_config_empty_string_config(self, graphql_context):
         assert result.data["isPipelineConfigValid"]["__typename"] == "RunConfigValidationInvalid"
         assert result.data["isPipelineConfigValid"]["pipelineName"] == "csv_hello_world"
 
+    def test_basic_valid_config_non_dict_config(self, graphql_context):
+        result = execute_config_graphql(
+            graphql_context,
+            pipeline_name="csv_hello_world",
+            run_config="daggy",
+            mode="default",
+        )
+
+        assert not result.errors
+        assert result.data
+        assert result.data["isPipelineConfigValid"]["__typename"] == "RunConfigValidationInvalid"
+        assert result.data["isPipelineConfigValid"]["pipelineName"] == "csv_hello_world"
+
     def test_root_field_not_defined(self, graphql_context):
         result = execute_config_graphql(
             graphql_context,