TemplateMapper creates new task or pipeline and maps properties to or…

…iginal if missing

src/engine/src/core/workflows/executors/WorkflowExecutor.py

@@ -209,7 +209,7 @@ def _prepare_tasks(self):
         objects, prepares the file system for each task execution, handles task templating,
         and generates and registers the task executors that will be called to perform the
         work detailed in the task definition."""
-
+        self.state.ctx.output = {}
         for task in self.state.ctx.pipeline.tasks:
             # Create an execution_uuid for each task in the pipeline
             task.execution_uuid = str(uuid4())
@@ -241,7 +241,7 @@ def _prepare_tasks(self):
                 task = template_mapper.map(task, task.uses)
 
             # Add a key to the output for the task
-            self.state.ctx.output = {task.id: []}
+            self.state.ctx.output[task.id] = None
 
             # Resolve and register the task executor
             executor = factory.build(task, self.state.ctx, self.exchange, plugins=self._plugins)
@@ -284,10 +284,10 @@ def _start_task(self, task):
 
                 self.state.ctx.output = {
                     **self.state.ctx.output,
-                    **task_result.output
+                    **{task.id: task_result.output}
                 }
             else:
-                task_result = TaskResult(0, data={"task": task.id})
+                task_result = TaskResult(0, output={task.id: None})
         except InvalidTaskTypeError as e:
             self.state.ctx.logger.error(str(e))
             task_result = TaskResult(1, errors=[str(e)])

src/engine/src/helpers/TemplateMapper.py

@@ -26,10 +26,28 @@ def __init__(self, cache_dir: str):
             "tapis_actor": TapisActorTask
         }
 
-    def map(self, obj: Union[Pipeline, Task], uses: Uses):
+    def map(self, obj: Union[Pipeline, Task], uses: Uses) -> Union[Pipeline, Task]:
+        """This method takes an object(Pipeline or Task object), and updates its
+        attributes with the values found in the templates
+        
+        IMPORTANT NOTE:
+        The original object is modified and returned by this function, not a copy.
+        """
+
         # Clone git repository specified on the pipeline.uses if exists
         template = TemplateRepository(uses, cache_dir=self.cache_dir)
 
+        # Resolve which class the final object should have
+        obj_class = Pipeline
+        if not issubclass(obj.__class__, Pipeline):
+            obj_class = self.task_map_by_type.get(obj.get("type"), None)
+
+        # Raise exception if no class could be resolved from the template
+        if obj_class == None:
+            raise Exception(f"Invalid Template: Unable to resolve object type from Template. Task template object 'type' property must be one of [{self.task_map_by_type.keys()}]")
+
+        dict_obj = obj.dict()
+
         for attr in template.keys():
             # For pipelines only. Skip the tasks property as they should be handled
             # seperately in another call to the map method of the Template ampper
@@ -39,10 +57,21 @@ def map(self, obj: Union[Pipeline, Task], uses: Uses):
             # For task only. The template should specify the correct type. For all other properties,
             # that are not specified, we use that which in enumerated in the template
             if attr == "type":
-                obj.type = template.get(attr)
+                dict_obj["type"] = template.get(attr)
+                continue
+
+            if obj.get(attr, None) == None:
+                dict_obj[attr] = template[attr]
+
+        new_obj = obj_class(**dict_obj)
+
+        for attr in vars(new_obj):
+            if attr == "tasks":
                 continue
 
-            if getattr(obj, attr) == None:
-                setattr(obj, attr, template[attr])
+            updated_value = getattr(new_obj, attr)
+            original_value = getattr(obj, attr)
+            if original_value != updated_value:
+                setattr(obj, attr, updated_value)
 
         return obj