workflows to jsonforms format

pyproject.toml

@@ -74,7 +74,7 @@ enabled = true
 
 [tool.pytest.ini_options]
 addopts = """--cov=omotes_rest --cov-report html --cov-report term-missing \
---cov-fail-under 2"""
+--cov-fail-under 40"""
 
 [tool.coverage.run]
 source = ["src"]

src/omotes_rest/apis/workflow.py

@@ -22,5 +22,5 @@ class JobAPI(MethodView):
     """Requests."""
 
     def get(self) -> Response:
-        """Return a summary of all jobs."""
-        return jsonify(current_app.rest_if.get_available_workflows_dict())
+        """Return a summary of all workflows with parameter jsonforms format."""
+        return jsonify(current_app.rest_if.get_workflows_jsonforms_format())

src/omotes_rest/rest_interface.py

@@ -1,6 +1,7 @@
 import base64
 import uuid
-from datetime import timedelta
+from datetime import timedelta, datetime
+from typing import Union
 
 from omotes_sdk.omotes_interface import OmotesInterface
 from omotes_sdk.internal.common.config import EnvRabbitMQConfig
@@ -9,9 +10,14 @@
     JobResult,
     JobProgressUpdate,
     JobStatusUpdate,
-    AvailableWorkflows,
 )
-from omotes_sdk.workflow_type import WorkflowTypeManager
+from omotes_sdk.workflow_type import (
+    StringParameter,
+    BooleanParameter,
+    IntegerParameter,
+    FloatParameter,
+    DateTimeParameter,
+)
 
 import logging
 from omotes_rest.postgres_interface import PostgresInterface
@@ -30,17 +36,12 @@ class RestInterface:
     """Interface to Omotes."""
     postgres_if: PostgresInterface
     """Interface to Omotes rest postgres."""
-    workflow_type_manager: WorkflowTypeManager
-    """Interface to Omotes."""
 
     def __init__(
         self,
     ) -> None:
         """Create the omotes rest interface."""
-        self.omotes_if = OmotesInterface(
-            EnvRabbitMQConfig(),
-            callback_on_available_workflows_update=self.handle_on_available_workflows_update,
-        )
+        self.omotes_if = OmotesInterface(EnvRabbitMQConfig())
         self.postgres_if = PostgresInterface(PostgresConfig())
 
     def start(self) -> None:
@@ -52,16 +53,6 @@ def stop(self) -> None:
         """Stop the omotes rest interface."""
         self.omotes_if.stop()
 
-    def handle_on_available_workflows_update(
-        self, available_workflows_pb: AvailableWorkflows
-    ) -> None:
-        """When the available workflows are updated.
-
-        :param available_workflows_pb: AvailableWorkflows protobuf message.
-        """
-        self.workflow_type_manager = WorkflowTypeManager.from_pb_message(available_workflows_pb)
-        logger.info("Updated the available workflows to: \n%s", self.workflow_type_manager)
-
     def handle_on_job_finished(self, job: Job, result: JobResult) -> None:
         """When a job is finished.
 
@@ -120,20 +111,79 @@ def handle_on_job_progress_update(self, job: Job, progress_update: JobProgressUp
             progress_message=progress_update.message,
         )
 
-    def get_available_workflows_dict(self) -> dict:
-        """Get the available workflows represented as dictionary.
+    def get_workflows_jsonforms_format(self) -> dict:
+        """Get the available workflows with jsonforms schemas per parameter.
 
         :return: dictionary response.
         """
-        return self.workflow_type_manager.to_dict()
+        workflows_dict = dict()
+        for _workflow in self.omotes_if.get_workflow_type_manager().get_all_workflows():
+            parameter_jsonforms_schemas = dict()
+            if _workflow.workflow_parameters:
+                for _parameter in _workflow.workflow_parameters:
+                    jsonforms_schema: dict[
+                        str, Union[str, float, datetime, list[dict[str, str]]]
+                    ] = dict()
+                    if _parameter.title:
+                        jsonforms_schema["title"] = _parameter.title
+                    if _parameter.description:
+                        jsonforms_schema["description"] = _parameter.description
+
+                    if isinstance(_parameter, StringParameter):
+                        jsonforms_schema["type"] = "string"
+                        if _parameter.default:
+                            jsonforms_schema["default"] = _parameter.default
+                        if _parameter.enum_options:
+                            one_of_list = []
+                            for enum_option in _parameter.enum_options:
+                                one_of_list.append(
+                                    dict(const=enum_option.key_name, title=enum_option.display_name)
+                                )
+                            jsonforms_schema["oneOf"] = one_of_list
+                    elif isinstance(_parameter, BooleanParameter):
+                        jsonforms_schema["type"] = "boolean"
+                        if _parameter.default:
+                            jsonforms_schema["default"] = _parameter.default
+                    elif isinstance(_parameter, IntegerParameter):
+                        jsonforms_schema["type"] = "integer"
+                        if _parameter.default:
+                            jsonforms_schema["default"] = _parameter.default
+                        if _parameter.minimum:
+                            jsonforms_schema["minimum"] = _parameter.minimum
+                        if _parameter.maximum:
+                            jsonforms_schema["maximum"] = _parameter.maximum
+                    elif isinstance(_parameter, FloatParameter):
+                        jsonforms_schema["type"] = "number"
+                        if _parameter.default:
+                            jsonforms_schema["default"] = _parameter.default
+                        if _parameter.minimum:
+                            jsonforms_schema["minimum"] = _parameter.minimum
+                        if _parameter.maximum:
+                            jsonforms_schema["maximum"] = _parameter.maximum
+                    elif isinstance(_parameter, DateTimeParameter):
+                        jsonforms_schema["type"] = "string"
+                        jsonforms_schema["format"] = "date-time"
+                        if _parameter.default:
+                            jsonforms_schema["default"] = _parameter.default.isoformat()
+                    else:
+                        raise NotImplementedError(
+                            f"Parameter type {type(_parameter)} not supported"
+                        )
+                    parameter_jsonforms_schemas[_parameter.key_name] = jsonforms_schema
+
+            workflows_dict[_workflow.workflow_type_name] = dict(
+                workflow_description=_workflow.workflow_type_description_name,
+                parameter_jsonforms_schemas=parameter_jsonforms_schemas,
+            )
+        return workflows_dict
 
     def submit_job(self, job_input: JobInput) -> JobStatusResponse:
         """When a job has a progress update.
 
         :param job_input: JobInput dataclass with job input.
         :return: JobStatusResponse.
         """
-        workflow_type = self.workflow_type_manager.get_workflow_by_name(
+        workflow_type = self.omotes_if.get_workflow_type_manager().get_workflow_by_name(
             FRONTEND_NAME_TO_OMOTES_WORKFLOW_NAME[job_input.workflow_type]
         )
         if not workflow_type:
@@ -183,7 +233,7 @@ def cancel_job(self, job_id: uuid.UUID) -> bool:
         job_in_db = self.get_job(job_id)
 
         if job_in_db:
-            workflow_type = self.workflow_type_manager.get_workflow_by_name(
+            workflow_type = self.omotes_if.get_workflow_type_manager().get_workflow_by_name(
                 FRONTEND_NAME_TO_OMOTES_WORKFLOW_NAME[job_in_db.workflow_type]
             )
             if not workflow_type: