fix: allow to create a complete version from a pipeline creation (HEXA-1661) (#1819)

Author: yolanfery

backend/hexa/mcp/tools/pipelines.py

@@ -257,18 +257,19 @@ def create_pipeline(
         return {"errors": [{"message": "source_code is required and cannot be empty"}]}
     zipfile_b64 = _build_zipfile_b64(source_code)
 
+    create_input = {
+        "workspaceSlug": workspace_slug,
+        "name": name,
+        "description": description or None,
+        "functionalType": functional_type or None,
+    }
+    if zipfile_b64 is not None:
+        create_input["version"] = {"zipfile": zipfile_b64}
+
     data = execute_graphql(
         user,
         "CreatePipeline",
-        {
-            "input": {
-                "workspaceSlug": workspace_slug,
-                "name": name,
-                "description": description or None,
-                "functionalType": functional_type or None,
-                "zipfile": zipfile_b64,
-            }
-        },
+        {"input": create_input},
     )
     if "errors" in data:
         return data

backend/hexa/pipelines/graphql/schema.graphql

@@ -421,6 +421,20 @@ enum PipelineOrderBy {
   LAST_RUN_DATE_ASC
 }
 
+"""
+Configures the first pipeline version, created atomically alongside the pipeline.
+Providing this sub-input signals that a first version should be created.
+"""
+input CreatePipelineVersionInput {
+  zipfile: String!  # Base64-encoded ZIP file containing the version code.
+  name: String  # The name of the first pipeline version.
+  description: String  # The description of the first pipeline version.
+  externalLink: URL  # The external link associated with the first pipeline version.
+  parameters: [ParameterInput!]  # The parameters of the first pipeline version (defaults to those parsed from the zipfile).
+  timeout: Int  # The timeout, in seconds, of the first pipeline version.
+  config: JSON  # The default configuration applied to runs of the first pipeline version.
+}
+
 """
 Represents the input for creating a pipeline.
 """
@@ -432,7 +446,7 @@ input CreatePipelineInput {
   notebookPath: String  # The path to the notebook file for the pipeline.
   functionalType: PipelineFunctionalType  # The functional type of the pipeline.
   tags: [String!]  # The names of tags to associate with the pipeline.
-  zipfile: String  # Optional base64-encoded ZIP file to upload as the first pipeline version.
+  version: CreatePipelineVersionInput  # Optional first pipeline version, created atomically with the pipeline.
 }
 
 """

backend/hexa/pipelines/schema/mutations.py

@@ -58,6 +58,23 @@ def _parse_parameters_from_zipfile(zipfile_data: bytes) -> list:
         raise PipelineCodeParsingError(str(e))
 
 
+def _validate_pipeline_version_timeout(
+    timeout: int | None, workspace: Workspace
+) -> None:
+    if not timeout:
+        return
+    max_allowed_timeout = int(settings.PIPELINE_RUN_MAX_TIMEOUT)
+    subscription = workspace and workspace.current_subscription
+    if subscription and subscription.max_pipeline_timeout:
+        max_allowed_timeout = min(
+            max_allowed_timeout, subscription.max_pipeline_timeout
+        )
+    if timeout < 0 or timeout > max_allowed_timeout:
+        raise InvalidTimeoutValueError(
+            "Pipeline timeout value cannot be negative or greater than the maximum allowed value."
+        )
+
+
 @pipelines_mutations.field("createPipeline")
 def resolve_create_pipeline(_, info, **kwargs):
     request: HttpRequest = info.context["request"]
@@ -106,14 +123,10 @@ def resolve_create_pipeline(_, info, **kwargs):
                 pipeline.tags.set(tags)
 
             version = None
-            if input.get("zipfile"):
-                zipfile_data = base64.b64decode(input["zipfile"].encode("ascii"))
-                parameters = _parse_parameters_from_zipfile(zipfile_data)
-
-                version = pipeline.upload_new_version(
-                    user=request.user,
-                    zipfile=zipfile_data,
-                    parameters=parameters,
+            version_input = input.get("version")
+            if version_input is not None:
+                version = _create_first_pipeline_version(
+                    request.user, workspace, pipeline, version_input
                 )
 
             event_properties = {
@@ -137,15 +150,42 @@ def resolve_create_pipeline(_, info, **kwargs):
             "errors": ["PIPELINE_CODE_PARSING_ERROR"],
             "details": str(e),
         }
+    except PipelineDoesNotSupportParametersError:
+        return {"success": False, "errors": ["PIPELINE_DOES_NOT_SUPPORT_PARAMETERS"]}
+    except InvalidTimeoutValueError:
+        return {"success": False, "errors": ["INVALID_TIMEOUT_VALUE"]}
 
     return {
         "pipeline": pipeline,
-        "pipelineVersion": version,
+        "pipeline_version": version,
         "success": True,
         "errors": [],
     }
 
 
+def _create_first_pipeline_version(
+    user: User, workspace: Workspace, pipeline: Pipeline, version_input: dict
+) -> PipelineVersion:
+    zipfile_data = base64.b64decode(version_input["zipfile"].encode("ascii"))
+    parameters = version_input.get("parameters") or _parse_parameters_from_zipfile(
+        zipfile_data
+    )
+
+    timeout = version_input.get("timeout")
+    _validate_pipeline_version_timeout(timeout, workspace)
+
+    return pipeline.upload_new_version(
+        user=user,
+        name=version_input.get("name"),
+        description=version_input.get("description"),
+        external_link=version_input.get("external_link"),
+        zipfile=zipfile_data,
+        parameters=parameters,
+        timeout=timeout,
+        config=version_input.get("config"),
+    )
+
+
 @pipelines_mutations.field("updatePipeline")
 def resolve_update_pipeline(_, info, **kwargs):
     request: HttpRequest = info.context["request"]
@@ -386,19 +426,7 @@ def resolve_upload_pipeline(_, info, **kwargs):
             "errors": ["PIPELINE_NOT_FOUND"],
         }
     try:
-        max_allowed_timeout = int(settings.PIPELINE_RUN_MAX_TIMEOUT)
-        subscription = pipeline.workspace and pipeline.workspace.current_subscription
-        if subscription and subscription.max_pipeline_timeout:
-            max_allowed_timeout = min(
-                max_allowed_timeout, subscription.max_pipeline_timeout
-            )
-
-        if input.get("timeout") and (
-            input.get("timeout") < 0 or input.get("timeout") > max_allowed_timeout
-        ):
-            raise InvalidTimeoutValueError(
-                "Pipeline timeout value cannot be negative or greater than the maximum allowed value."
-            )
+        _validate_pipeline_version_timeout(input.get("timeout"), pipeline.workspace)
 
         zipfile_data = base64.b64decode(input.get("zipfile").encode("ascii"))
         parameters = input.get("parameters")

backend/hexa/pipelines/tests/test_schema/test_pipelines.py

@@ -252,6 +252,144 @@ def test_create_pipeline_notebook(self):
             pipeline = Pipeline.objects.filter_for_user(self.USER_ROOT).get()
             self.assertEqual(pipeline.type, PipelineType.NOTEBOOK)
 
+    CREATE_PIPELINE_WITH_VERSION_MUTATION = """
+        mutation createPipeline($input: CreatePipelineInput!) {
+            createPipeline(input: $input) {
+                success
+                errors
+                pipeline { code name }
+                pipelineVersion { name parameters { code } }
+            }
+        }
+    """
+
+    def test_create_pipeline_with_first_version(self):
+        self.assertEqual(0, len(Pipeline.objects.all()))
+        self.client.force_login(self.USER_ROOT)
+
+        r = self.run_query(
+            self.CREATE_PIPELINE_WITH_VERSION_MUTATION,
+            {
+                "input": {
+                    "name": "AtomicPipeline",
+                    "workspaceSlug": self.WS1.slug,
+                    "version": {
+                        "zipfile": self.zip_pipeline_py,
+                        "name": "v1",
+                    },
+                }
+            },
+        )
+
+        result = r["data"]["createPipeline"]
+        self.assertTrue(result["success"])
+        self.assertEqual(result["errors"], [])
+        self.assertEqual(result["pipeline"]["name"], "AtomicPipeline")
+        self.assertEqual(result["pipelineVersion"]["name"], "v1")
+        self.assertEqual(
+            [p["code"] for p in result["pipelineVersion"]["parameters"]],
+            ["file_path"],
+        )
+        pipeline = Pipeline.objects.filter_for_user(self.USER_ROOT).get()
+        self.assertEqual(pipeline.versions.count(), 1)
+
+    def test_create_pipeline_with_first_version_negative_timeout(self):
+        self.assertEqual(0, len(Pipeline.objects.all()))
+        self.client.force_login(self.USER_ROOT)
+
+        r = self.run_query(
+            self.CREATE_PIPELINE_WITH_VERSION_MUTATION,
+            {
+                "input": {
+                    "name": "AtomicPipeline",
+                    "workspaceSlug": self.WS1.slug,
+                    "version": {
+                        "zipfile": self.zip_pipeline_py,
+                        "timeout": -1,
+                    },
+                }
+            },
+        )
+
+        self.assertEqual(
+            r["data"]["createPipeline"]["errors"], ["INVALID_TIMEOUT_VALUE"]
+        )
+        self.assertFalse(r["data"]["createPipeline"]["success"])
+        self.assertEqual(0, Pipeline.objects.all().count())
+
+    def test_create_pipeline_with_first_version_timeout_too_high(self):
+        self.assertEqual(0, len(Pipeline.objects.all()))
+        self.client.force_login(self.USER_ROOT)
+
+        r = self.run_query(
+            self.CREATE_PIPELINE_WITH_VERSION_MUTATION,
+            {
+                "input": {
+                    "name": "AtomicPipeline",
+                    "workspaceSlug": self.WS1.slug,
+                    "version": {
+                        "zipfile": self.zip_pipeline_py,
+                        "timeout": int(settings.PIPELINE_RUN_MAX_TIMEOUT) + 1,
+                    },
+                }
+            },
+        )
+
+        self.assertEqual(
+            r["data"]["createPipeline"]["errors"], ["INVALID_TIMEOUT_VALUE"]
+        )
+        self.assertFalse(r["data"]["createPipeline"]["success"])
+        self.assertEqual(0, Pipeline.objects.all().count())
+
+    def test_create_pipeline_with_first_version_bad_zipfile(self):
+        self.assertEqual(0, len(Pipeline.objects.all()))
+        self.client.force_login(self.USER_ROOT)
+
+        bad_zip = base64.b64encode(b"not a real zip").decode("ascii")
+        r = self.run_query(
+            self.CREATE_PIPELINE_WITH_VERSION_MUTATION,
+            {
+                "input": {
+                    "name": "AtomicPipeline",
+                    "workspaceSlug": self.WS1.slug,
+                    "version": {"zipfile": bad_zip},
+                }
+            },
+        )
+
+        self.assertEqual(
+            r["data"]["createPipeline"]["errors"], ["PIPELINE_CODE_PARSING_ERROR"]
+        )
+        self.assertFalse(r["data"]["createPipeline"]["success"])
+        self.assertEqual(0, Pipeline.objects.all().count())
+
+    def test_create_pipeline_with_first_version_explicit_parameters(self):
+        self.assertEqual(0, len(Pipeline.objects.all()))
+        self.client.force_login(self.USER_ROOT)
+
+        r = self.run_query(
+            self.CREATE_PIPELINE_WITH_VERSION_MUTATION,
+            {
+                "input": {
+                    "name": "AtomicPipeline",
+                    "workspaceSlug": self.WS1.slug,
+                    "version": {
+                        "zipfile": self.zip_pipeline_py,
+                        "parameters": pipelines_parameters_with_scalars,
+                    },
+                }
+            },
+        )
+
+        self.assertTrue(r["data"]["createPipeline"]["success"])
+        self.assertEqual(
+            [
+                p["code"]
+                for p in r["data"]["createPipeline"]["pipelineVersion"]["parameters"]
+            ],
+            [p["code"] for p in pipelines_parameters_with_scalars],
+        )
+
     def test_list_pipelines(self):
         self.assertEqual(0, len(PipelineRun.objects.all()))
         self._create_pipeline(name="Pipeline DHIS2")

frontend/schema.generated.graphql

@@ -1042,8 +1042,8 @@ input CreatePipelineInput {
   name: String!
   notebookPath: String
   tags: [String!]
+  version: CreatePipelineVersionInput
   workspaceSlug: String!
-  zipfile: String
 }
 
 """Represents the input for adding a recipient to a pipeline."""
@@ -1095,6 +1095,20 @@ type CreatePipelineTemplateVersionResult {
   success: Boolean!
 }
 
+"""
+Configures the first pipeline version, created atomically alongside the pipeline.
+Providing this sub-input signals that a first version should be created.
+"""
+input CreatePipelineVersionInput {
+  config: JSON
+  description: String
+  externalLink: URL
+  name: String
+  parameters: [ParameterInput!]
+  timeout: Int
+  zipfile: String!
+}
+
 """
 The CreateTeamError enum represents the possible errors that can occur during the createTeam mutation.
 """

frontend/src/graphql/types.ts

@@ -1053,8 +1053,8 @@ export type CreatePipelineInput = {
   name: Scalars['String']['input'];
   notebookPath?: InputMaybe<Scalars['String']['input']>;
   tags?: InputMaybe<Array<Scalars['String']['input']>>;
+  version?: InputMaybe<CreatePipelineVersionInput>;
   workspaceSlug: Scalars['String']['input'];
-  zipfile?: InputMaybe<Scalars['String']['input']>;
 };
 
 /** Represents the input for adding a recipient to a pipeline. */
@@ -1106,6 +1106,20 @@ export type CreatePipelineTemplateVersionResult = {
   success: Scalars['Boolean']['output'];
 };
 
+/**
+ * Configures the first pipeline version, created atomically alongside the pipeline.
+ * Providing this sub-input signals that a first version should be created.
+ */
+export type CreatePipelineVersionInput = {
+  config?: InputMaybe<Scalars['JSON']['input']>;
+  description?: InputMaybe<Scalars['String']['input']>;
+  externalLink?: InputMaybe<Scalars['URL']['input']>;
+  name?: InputMaybe<Scalars['String']['input']>;
+  parameters?: InputMaybe<Array<ParameterInput>>;
+  timeout?: InputMaybe<Scalars['Int']['input']>;
+  zipfile: Scalars['String']['input'];
+};
+
 /** The CreateTeamError enum represents the possible errors that can occur during the createTeam mutation. */
 export enum CreateTeamError {
   /** Indicates that a team with the same name already exists. */