Merge pull request #54 from wayfair-incubator/reprocess

Allowing re-execution of tasks based on a reprocessing the message

Author: patkivikram

dagger/tasks/task.py

@@ -322,15 +322,21 @@ async def start(
         ignore_status: bool = False,
     ) -> None:
         # pre-execute
-        if self.status.code in [
-            TaskStatusEnum.COMPLETED.name,
-            TaskStatusEnum.SKIPPED.name,
-        ]:
+        if (
+            self.status.code
+            in [
+                TaskStatusEnum.COMPLETED.name,
+                TaskStatusEnum.SKIPPED.name,
+            ]
+            and not self.reprocess_on_message
+        ):
             return await self.on_complete(
                 status=self.status, workflow_instance=workflow_instance
             )
         if (
-            ignore_status or self.status.code == TaskStatusEnum.NOT_STARTED.name
+            ignore_status
+            or self.status.code == TaskStatusEnum.NOT_STARTED.name
+            or self.reprocess_on_message
         ) and workflow_instance:
             self.status = TaskStatus(
                 code=TaskStatusEnum.EXECUTING.name, value=TaskStatusEnum.EXECUTING.value
@@ -964,16 +970,21 @@ async def stop(
         )
 
     async def start(self, workflow_instance: Optional[ITemplateDAGInstance]) -> None:
-        if self.status.code in [
-            TaskStatusEnum.COMPLETED.name,
-            TaskStatusEnum.SKIPPED.name,
-        ]:
+        if (
+            self.status.code
+            in [
+                TaskStatusEnum.COMPLETED.name,
+                TaskStatusEnum.SKIPPED.name,
+            ]
+            and not self.reprocess_on_message
+        ):
             return await self.on_complete(
                 status=self.status, workflow_instance=workflow_instance
             )
         if (
             self.status.code == TaskStatusEnum.NOT_STARTED.name
             or self.status.code == TaskStatusEnum.SUBMITTED.name
+            or self.reprocess_on_message
         ) and workflow_instance:
             await self.execute(
                 runtime_parameters=workflow_instance.runtime_parameters,