add GET workflow/{id}

Author: andrii-i

jupyter_scheduler/extension.py

@@ -38,6 +38,7 @@ class SchedulerApp(ExtensionApp):
         (r"scheduler/runtime_environments", RuntimeEnvironmentsHandler),
         (r"scheduler/config", ConfigHandler),
         (r"scheduler/worklows", WorkflowHandler),
+        (r"scheduler/worklows/{}".format(WORKFLOW_ID_REGEX), WorkflowHandler),
         (
             fr"scheduler/worklows/{WORKFLOW_ID_REGEX}/run",
             WorkflowRunHandler,

jupyter_scheduler/models.py

@@ -42,7 +42,6 @@ def __str__(self) -> str:
 
 
 class Status(str, Enum):
-    DRAFT = "DRAFT"
     CREATED = "CREATED"
     QUEUED = "QUEUED"
     IN_PROGRESS = "IN_PROGRESS"

jupyter_scheduler/orm.py

@@ -114,7 +114,7 @@ class Workflow(Base):
     __table_args__ = {"extend_existing": True}
     workflow_id = Column(String(36), primary_key=True, default=generate_uuid)
     tasks = Column(JsonType(1024))
-    status = Column(String(64), default=Status.DRAFT)
+    status = Column(String(64), default=Status.CREATED)
     # All new columns added to this table must be nullable to ensure compatibility during database migrations.
     # Any default values specified for new columns will be ignored during the migration process.
 

jupyter_scheduler/scheduler.py

@@ -46,7 +46,7 @@
     create_output_directory,
     create_output_filename,
 )
-from jupyter_scheduler.workflows import CreateWorkflow
+from jupyter_scheduler.workflows import CreateWorkflow, DescribeWorkflow
 
 
 class BaseScheduler(LoggingConfigurable):
@@ -113,7 +113,15 @@ def create_job(self, model: CreateJob) -> str:
         raise NotImplementedError("must be implemented by subclass")
 
     def create_workflow(self, model: CreateWorkflow) -> str:
-        """Creates a new workflow record, may trigger execution of the workflow."""
+        """Creates a new workflow record."""
+        raise NotImplementedError("must be implemented by subclass")
+
+    def run_workflow(self, workflow_id: str) -> str:
+        """Triggers execution of the workflow."""
+        raise NotImplementedError("must be implemented by subclass")
+
+    def get_workflow(self, workflow_id: str) -> DescribeWorkflow:
+        """Returns workflow record for a single workflow."""
         raise NotImplementedError("must be implemented by subclass")
 
     def update_job(self, job_id: str, model: UpdateJob):
@@ -546,6 +554,14 @@ def run_workflow(self, workflow_id: str) -> str:
         execution_manager.process_workflow()
         return workflow_id
 
+    def get_workflow(self, workflow_id: str) -> DescribeWorkflow:
+        with self.db_session() as session:
+            workflow_record = (
+                session.query(Workflow).filter(Workflow.workflow_id == workflow_id).one()
+            )
+        model = DescribeWorkflow.from_orm(workflow_record)
+        return model
+
     def update_job(self, job_id: str, model: UpdateJob):
         with self.db_session() as session:
             session.query(Job).filter(Job.job_id == job_id).update(model.dict(exclude_none=True))

jupyter_scheduler/workflows.py

@@ -44,6 +44,21 @@ async def post(self):
         else:
             self.finish(json.dumps(dict(workflow_id=workflow_id)))
 
+    @authenticated
+    async def get(self, workflow_id: str = None):
+        if not workflow_id:
+            raise HTTPError(400, "Missing workflow_id in the request URL.")
+        try:
+            workflow = await ensure_async(self.scheduler.get_workflow(workflow_id))
+        except SchedulerError as e:
+            self.log.exception(e)
+            raise HTTPError(500, str(e)) from e
+        except Exception as e:
+            self.log.exception(e)
+            raise HTTPError(500, "Unexpected error occurred while getting workflow details.") from e
+        else:
+            self.finish(workflow.json())
+
 
 class WorkflowRunHandler(ExtensionHandlerMixin, JobHandlersMixin, APIHandler):
     @authenticated