feat: API adapter now handles tasks

Author: Alan Christie

tests/api_adapter.py

@@ -33,6 +33,7 @@ def __init__(self):
         self._running_workflow: Dict[str, Dict[str, Any]] = {}
         self._running_workflow_steps: Dict[str, Dict[str, Any]] = {}
         self._instances: Dict[str, Dict[str, Any]] = {}
+        self._tasks: Dict[str, Dict[str, Any]] = {}
 
     def create_workflow(self, *, workflow_definition: Dict[str, Any]) -> str:
         next_id: int = len(self._workflow_definitions) + 1
@@ -108,13 +109,29 @@ def create_instance(self, *, running_workflow_step_id: str) -> Dict[str, Any]:
             "running_workflow_step": running_workflow_step_id,
         }
         self._instances[instance_id] = record
-        return {"instance_id": running_workflow_step_id}
+        return {"id": running_workflow_step_id}
 
     def get_instance(self, *, instance_id: str) -> Dict[str, Any]:
         if instance_id not in self._instances:
             return {}
         return {self._instances[instance_id]}
 
+    def create_task(self, *, instance_id: str) -> Dict[str, Any]:
+        next_id: int = len(self._instances) + 1
+        task_id: str = _INSTANCE_ID_FORMAT.format(id=next_id)
+        record = {
+            "instance_id": instance_id,
+            "done": False,
+            "exit_code": 0,
+        }
+        self._tasks[task_id] = record
+        return {"id": task_id}
+
+    def get_task(self, *, task_id: str) -> Dict[str, Any]:
+        if task_id not in self._tasks:
+            return {}
+        return {self._tasks[task_id]}
+
     def get_job(
         self, *, collection: str, job: str, version: str
     ) -> Optional[Dict[str, Any]]:

tests/instance_launcher.py

@@ -47,11 +47,13 @@ def launch(
             running_workflow_step_id=running_workflow_step_id
         )
         assert "running_workflow_step" in response
-        # Now simulate the creation of an Instance record
+        # Now simulate the creation of a Task and Instance record
         response = self._api_adapter.create_instance(
             running_workflow_step_id=running_workflow_step_id
         )
-        instance_id = response["instance_id"]
+        instance_id = response["id"]
+        response = self._api_adapter.create_task(instance_id=instance_id)
+        task_id = response["id"]
 
         # Just run the Python module that matched the 'job' in the step specification.
         # Don't care about 'version' or 'collection'.
@@ -70,14 +72,16 @@ def launch(
         pod_message.timestamp = f"{datetime.now(timezone.utc).isoformat()}Z"
         pod_message.phase = "Completed"
         pod_message.instance = instance_id
+        pod_message.task = task_id
         pod_message.has_exit_code = True
         pod_message.exit_code = 0
         self._msg_dispatcher.send(pod_message)
 
         return LaunchResult(
+            # The errors returned here are the launch errors, not the Job's errors.
             error=0,
             error_msg=None,
             instance_id=instance_id,
-            task_id="task-00000000-0000-0000-0000-000000000000",
+            task_id=task_id,
             command=" ".join(job_cmd),
         )

tests/test_test_instance_launcher.py

@@ -38,7 +38,6 @@ def test_get_nop_job(basic_launcher):
         workflow_id="workflow-00000000-0000-0000-0000-000000000000",
         running_workflow_step_id=rwfsid,
         workflow_definition={},
-        step="step-1",
         step_specification={"job": "nop"},
     )
 

workflow/workflow_abc.py

@@ -181,6 +181,24 @@ def get_instance(self, *, instance_id: str) -> Dict[str, Any]:
         #    [...],
         # }
 
+    @abstractmethod
+    def create_task(self, instance_id: str) -> Dict[str, Any]:
+        """Create a Task Record (for amn Instance)"""
+        # Should return:
+        # {
+        #    "id": "task-00000000-0000-0000-0000-000000000001",
+        # }
+
+    @abstractmethod
+    def get_task(self, *, task_id: str) -> Dict[str, Any]:
+        """Get a Task Record"""
+        # Should return:
+        # {
+        #    "done": True,
+        #    "exit_code": 0,
+        #    [...],
+        # }
+
     @abstractmethod
     def get_job(
         self,