Merge

hexa/pipeline_templates/graphql/schema.graphql

@@ -3,6 +3,7 @@ extend type Mutation {
   Creates a new pipeline template version.
   """
   createPipelineTemplateVersion(input: CreatePipelineTemplateVersionInput!): CreatePipelineTemplateVersionResult! @loginRequired
+  createPipelineFromTemplateVersion(input: CreatePipelineFromTemplateVersionInput!): CreatePipelineFromTemplateVersionResult! @loginRequired
 }
 
 extend type Query {
@@ -40,6 +41,33 @@ type CreatePipelineTemplateVersionResult {
   errors: [CreatePipelineTemplateVersionError!]  # The list of errors that occurred during the creation of the pipeline template version.
 }
 
+"""
+Represents the input for creating a new pipeline from a template version.
+"""
+input CreatePipelineFromTemplateVersionInput {
+  workspaceSlug: String!  # The slug of the pipeline workspace.
+  pipelineTemplateVersionId: UUID!  # The ID of the pipeline template version.
+}
+
+"""
+Represents the result of creating a new pipeline from a template version.
+"""
+type CreatePipelineFromTemplateVersionResult {
+  pipeline: Pipeline  # The created pipeline.
+  success: Boolean!  # Indicates if the pipeline was created successfully.
+  errors: [CreatePipelineFromTemplateVersionError!]  # The list of errors that occurred during the creation of the pipeline.
+}
+
+"""
+Enum representing the possible errors that can occur when creating a pipeline from a template version.
+"""
+enum CreatePipelineFromTemplateVersionError {
+  PERMISSION_DENIED
+  WORKSPACE_NOT_FOUND
+  PIPELINE_TEMPLATE_VERSION_NOT_FOUND
+  PIPELINE_ALREADY_EXISTS
+}
+
 """
 Enum representing the possible errors that can occur when creating a pipeline template version.
 """

hexa/pipeline_templates/models.py

@@ -100,5 +100,26 @@ class Meta:
 
     objects = PipelineTemplateVersionQuerySet.as_manager()
 
+    def create_pipeline(self, code, workspace, user):
+        source_pipeline = self.template.source_pipeline
+        source_version = self.source_pipeline_version
+        pipeline = Pipeline.objects.create(
+            source_template=self.template,
+            code=code,
+            name=source_pipeline.name,
+            description=source_pipeline.description,
+            config=source_pipeline.config,
+            workspace=workspace,
+        )
+        PipelineVersion.objects.create(
+            user=user,
+            pipeline=pipeline,
+            zipfile=source_version.zipfile,
+            parameters=source_version.parameters,
+            config=source_version.config,
+            timeout=source_version.timeout,
+        )
+        return pipeline
+
     def __str__(self):
         return f"v{self.version_number} of {self.template.name}"

hexa/pipeline_templates/schema/mutations.py

@@ -2,6 +2,7 @@
 from django.http import HttpRequest
 
 from hexa.analytics.api import track
+from hexa.pipeline_templates.models import PipelineTemplateVersion
 from hexa.pipelines.models import Pipeline, PipelineVersion
 from hexa.workspaces.models import Workspace
 
@@ -74,4 +75,40 @@ def resolve_create_pipeline_template_version(_, info, **kwargs):
     return {"pipeline_template": pipeline_template, "success": True, "errors": []}
 
 
+@pipeline_template_mutations.field("createPipelineFromTemplateVersion")
+def resolve_create_pipeline_from_template_version(_, info, **kwargs):
+    request: HttpRequest = info.context["request"]
+    input = kwargs["input"]
+
+    workspace = get_workspace(request.user, input.get("workspace_slug"))
+    if not workspace:
+        return {"success": False, "errors": ["WORKSPACE_NOT_FOUND"]}
+
+    if not request.user.has_perm("pipelines.create_pipeline", workspace):
+        return {"success": False, "errors": ["PERMISSION_DENIED"]}
+
+    try:
+        template_version = PipelineTemplateVersion.objects.get(
+            id=input["pipeline_template_version_id"]
+        )
+    except PipelineTemplateVersion.DoesNotExist:
+        return {"success": False, "errors": ["PIPELINE_TEMPLATE_VERSION_NOT_FOUND"]}
+
+    pipeline_code = f"{template_version.template.source_pipeline.code} (from Template)"
+    if Pipeline.objects.filter(workspace=workspace, code=pipeline_code).exists():
+        return {"success": False, "errors": ["PIPELINE_ALREADY_EXISTS"]}
+    pipeline = template_version.create_pipeline(pipeline_code, workspace, request.user)
+
+    track(
+        request,
+        "pipeline_templates.pipeline_created_from_template",
+        {
+            "pipeline_id": str(pipeline.id),
+            "template_version_id": str(template_version.id),
+            "workspace": workspace.slug,
+        },
+    )
+    return {"pipeline": pipeline, "success": True, "errors": []}
+
+
 bindables = [pipeline_template_mutations]

hexa/pipeline_templates/tests/test_schema/test_templates.py

@@ -33,11 +33,16 @@ def setUpTestData(cls):
                 name="WS1",
                 description="Workspace 1",
             )
-        cls.PIPELINE = Pipeline.objects.create(name="Test Pipeline", workspace=cls.WS1)
+        cls.PIPELINE = Pipeline.objects.create(
+            name="Test Pipeline", code="Test Pipeline", workspace=cls.WS1
+        )
         cls.PIPELINE_VERSION1 = PipelineVersion.objects.create(
             pipeline=cls.PIPELINE,
             version_number=1,
             description="Initial version",
+            parameters=[{"code": "param_1"}],
+            config=[{"param_1": 1}],
+            zipfile=str.encode("some_bytes"),
         )
         cls.PIPELINE_VERSION2 = PipelineVersion.objects.create(
             pipeline=cls.PIPELINE,
@@ -89,6 +94,69 @@ def test_create_template_version(self):
             self.PIPELINE_VERSION2.id, [{"versionNumber": 1}, {"versionNumber": 2}]
         )
 
+    def test_create_pipeline_from_template_version(self):
+        self.client.force_login(self.USER_ROOT)
+        self.create_template_version(self.PIPELINE_VERSION1.id, [{"versionNumber": 1}])
+        r = self.run_query(
+            """
+                mutation createPipelineFromTemplateVersion($input: CreatePipelineFromTemplateVersionInput!) {
+                    createPipelineFromTemplateVersion(input: $input) {
+                        success errors pipeline {name code currentVersion {zipfile parameters {code default} config}}
+                    }
+                }
+            """,
+            {
+                "input": {
+                    "workspaceSlug": self.WS1.slug,
+                    "pipelineTemplateVersionId": str(
+                        self.PIPELINE_VERSION1.template_version.id
+                    ),
+                }
+            },
+        )
+        self.assertEqual(
+            {
+                "success": True,
+                "errors": [],
+                "pipeline": {
+                    "name": self.PIPELINE.name,
+                    "code": "Test Pipeline (from Template)",
+                    "currentVersion": {
+                        "zipfile": "c29tZV9ieXRlcw==",
+                        "parameters": [{"code": "param_1", "default": None}],
+                        "config": [{"param_1": 1}],
+                    },
+                },
+            },
+            r["data"]["createPipelineFromTemplateVersion"],
+        )
+
+        r = self.run_query(
+            """
+                mutation createPipelineFromTemplateVersion($input: CreatePipelineFromTemplateVersionInput!) {
+                    createPipelineFromTemplateVersion(input: $input) {
+                        success errors pipeline {name code currentVersion {zipfile parameters {code default} config}}
+                    }
+                }
+            """,
+            {
+                "input": {
+                    "workspaceSlug": self.WS1.slug,
+                    "pipelineTemplateVersionId": str(
+                        self.PIPELINE_VERSION1.template_version.id
+                    ),
+                }
+            },
+        )
+        self.assertEqual(
+            {
+                "success": False,
+                "errors": ["PIPELINE_ALREADY_EXISTS"],
+                "pipeline": None,
+            },
+            r["data"]["createPipelineFromTemplateVersion"],
+        )
+
     def test_get_pipeline_templates(self):
         PipelineTemplate.objects.create(
             name="Template 1", code="Code 1", source_pipeline=self.PIPELINE

hexa/pipelines/migrations/0054_pipeline_source_template.py

@@ -0,0 +1,25 @@
+# Generated by Django 4.2.17 on 2024-12-20 16:26
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("pipeline_templates", "0002_alter_pipelinetemplate_name"),
+        ("pipelines", "0053_pipelinerun_log_level"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="pipeline",
+            name="source_template",
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name="pipelines",
+                to="pipeline_templates.pipelinetemplate",
+            ),
+        ),
+    ]

hexa/pipelines/models.py

@@ -293,6 +293,13 @@ class Meta:
         default=PipelineType.ZIPFILE,
     )
     notebook_path = models.TextField(null=True, blank=True)
+    source_template = models.ForeignKey(
+        "pipeline_templates.PipelineTemplate",
+        null=True,
+        blank=True,
+        on_delete=models.SET_NULL,
+        related_name="pipelines",
+    )
 
     objects = DefaultSoftDeletedManager.from_queryset(PipelineQuerySet)()
     all_objects = IncludeSoftDeletedManager.from_queryset(PipelineQuerySet)()
@@ -341,6 +348,40 @@ def is_schedulable(self):
         elif self.type == PipelineType.ZIPFILE:
             return self.last_version and self.last_version.is_schedulable
 
+    def get_config_from_previous_version(self, new_parameters: dict):
+        """
+        Get the config from the previous version of the pipeline considering only overlapping parameters between the new and the previous version.
+        """
+        previous_config_from_overlapping_parameters = {}
+        if self.last_version:
+            previous_parameters = self.last_version.parameters
+            overlapping_parameters = [
+                new_parameter
+                for new_parameter in new_parameters
+                if new_parameter in previous_parameters
+            ]
+            previous_config_from_overlapping_parameters = {
+                overlapping_parameter["code"]: value
+                for overlapping_parameter in overlapping_parameters
+                if (
+                    value := self.last_version.config.get(
+                        overlapping_parameter["code"],
+                        overlapping_parameter.get("default"),
+                    )
+                )
+                is not None
+            }
+        return {
+            new_parameter["code"]: value
+            for new_parameter in new_parameters
+            if (
+                value := previous_config_from_overlapping_parameters.get(
+                    new_parameter["code"], new_parameter.get("default")
+                )
+            )
+            is not None
+        }
+
     def upload_new_version(
         self,
         user: User,
@@ -355,14 +396,8 @@ def upload_new_version(
         if not user.has_perm("pipelines.update_pipeline", self):
             raise PermissionDenied
 
-        if config is None:
-            # No default configuration has been provided, let's take the default values from the parameters
-            # In the future, we'll use the one from the last version
-            config = {
-                parameter["code"]: parameter["default"]
-                for parameter in parameters
-                if parameter.get("default") is not None
-            }
+        config = config or self.get_config_from_previous_version(parameters)
+
         version = PipelineVersion(
             user=user,
             pipeline=self,