FWF-4504: [Feature] Task outcome configuration model and APIs

Author: auslin-aot

forms-flow-api/migrations/versions/ac75256dc5c9_task_outcome_configuration_table.py renamed to forms-flow-api/migrations/versions/716e21e9864f_task_outcome_configuration_table.py

@@ -1,16 +1,16 @@
 """Task_outcome_configuration_table
 
-Revision ID: 47d590c1d289
+Revision ID: 716e21e9864f
 Revises: a5d9bbf7b5ac
-Create Date: 2025-05-05 17:50:42.349349
+Create Date: 2025-05-06 16:50:10.473682
 
 """
 from alembic import op
 import sqlalchemy as sa
-from sqlalchemy.dialects import postgresql
+
 
 # revision identifiers, used by Alembic.
-revision = 'ac75256dc5c9'
+revision = '716e21e9864f'
 down_revision = 'a5d9bbf7b5ac'
 branch_labels = None
 depends_on = None
@@ -21,8 +21,10 @@ def upgrade():
     op.create_table('task_outcome_configuration',
     sa.Column('id', sa.Integer(), nullable=False),
     sa.Column('task_id', sa.String(length=100), nullable=False, comment='Task ID'),
-    sa.Column('task_outcome', postgresql.ARRAY(sa.JSON()), nullable=True, comment='Task outcome'),
-    sa.Column('created_by', sa.String(length=100), nullable=True, comment='Created by'),
+    sa.Column('task_name', sa.String(length=100), nullable=True, comment='Task name'),
+    sa.Column('task_transition_map', sa.JSON(), nullable=True, comment='Task transition map'),
+    sa.Column('transition_map_type', sa.String(length=100), nullable=True, comment='Task transition map type'),
+    sa.Column('created_by', sa.String(length=100), nullable=False, comment='Created by'),
     sa.Column('tenant', sa.String(length=100), nullable=True, comment='Tenant key'),
     sa.Column('created', sa.DateTime(timezone=True), nullable=False),
     sa.PrimaryKeyConstraint('id')

forms-flow-api/src/formsflow_api/constants/__init__.py

@@ -138,6 +138,10 @@ class BusinessErrorCode(ErrorCodeMixin, Enum):
         "Database error while updating filter preferences",
         HTTPStatus.BAD_REQUEST,
     )
+    TASK_OUTCOME_NOT_FOUND = (
+        "Task outcome configuration not found for the given task Id",
+        HTTPStatus.BAD_REQUEST,
+    )
 
     def __new__(cls, message, status_code):
         """Constructor."""

forms-flow-api/src/formsflow_api/models/tasks.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 from sqlalchemy import JSON
-from sqlalchemy.dialects.postgresql import ARRAY
 
 from .audit_mixin import ApplicationAuditDateTimeMixin
 from .base_model import BaseModel
@@ -17,27 +16,33 @@ class TaskOutcomeConfiguration(ApplicationAuditDateTimeMixin, BaseModel, db.Mode
     task_id = db.Column(
         db.String(100), nullable=False, comment="Task ID", unique=True, index=True
     )
-    task_outcome = db.Column(ARRAY(JSON), nullable=True, comment="Task outcome")
-    created_by = db.Column(db.String(100), nullable=True, comment="Created by")
+    task_name = db.Column(db.String(100), nullable=True, comment="Task name")
+    task_transition_map = db.Column(JSON, nullable=True, comment="Task transition map")
+    transition_map_type = db.Column(
+        db.String(100), nullable=True, comment="Task transition map type"
+    )
+    created_by = db.Column(db.String(100), nullable=False, comment="Created by")
     tenant = db.Column(db.String(100), nullable=True, comment="Tenant key", index=True)
 
     @classmethod
     def create_from_dict(
         cls, task_outcome_info: dict
     ) -> TaskOutcomeConfiguration | None:
-        """Create new task outcome."""
+        """Create new task outcome configuration."""
         if task_outcome_info:
             task = TaskOutcomeConfiguration()
             task.created_by = task_outcome_info.get("created_by")
             task.task_id = task_outcome_info.get("task_id")
-            task.task_outcome = task_outcome_info.get("task_outcome")
+            task.task_name = task_outcome_info.get("task_name")
+            task.task_transition_map = task_outcome_info.get("task_transition_map")
+            task.transition_map_type = task_outcome_info.get("transition_map_type")
             task.tenant = task_outcome_info.get("tenant")
             task.save()
             return task
         return None
 
     @classmethod
-    def get_task_outcome_by_task_id(
+    def get_task_outcome_configuration_by_task_id(
         cls, task_id: str, tenant: str
     ) -> TaskOutcomeConfiguration | None:
         """Get task outcome configuration by task ID."""

forms-flow-api/src/formsflow_api/resources/tasks.py

@@ -18,20 +18,23 @@
     "TaskOutcomeRequest",
     {
         "taskId": fields.String(description="Task ID", required=True),
-        "taskOutcome": fields.List(
-            fields.Raw(description="Task outcome"), required=True, min_items=1
+        "taskName": fields.String(
+            description="Task name", required=True, allow_none=True
+        ),
+        "transitionMapType": fields.String(
+            description="Task transition map type - select/input/radio", required=True
+        ),
+        "taskTransitionMap": fields.Raw(
+            description="Determines the next step in workflow", required=True
         ),
     },
 )
 
-task_outcome_response = API.model(
+task_outcome_response = API.inherit(
     "TaskOutcomeResponse",
+    task_outcome_request,
     {
-        "id": fields.Integer(description="Task outcome ID"),
-        "taskId": fields.String(description="Task ID"),
-        "taskOutcome": fields.List(
-            fields.Raw(description="Task outcome"), required=True, min_items=1
-        ),
+        "id": fields.Integer(description="Task outcome configuration ID"),
         "createdBy": fields.String(description="Created by"),
         "tenant": fields.String(description="Tenant key"),
         "created": fields.DateTime(description="Created date"),
@@ -40,7 +43,7 @@
 
 
 @cors_preflight("POST, OPTIONS")
-@API.route("/task-outcome", methods=["POST", "OPTIONS"])
+@API.route("/task-outcome-configuration", methods=["POST", "OPTIONS"])
 class TaskOutcomeResource(Resource):
     """Resource to create task outcome configuration."""
 
@@ -60,12 +63,12 @@ def post():
         data = request.get_json()
         if not data:
             return {"message": "Invalid input"}, HTTPStatus.BAD_REQUEST
-        response = TaskService().create_task_outcome(data)
+        response = TaskService().create_task_outcome_configuration(data)
         return response, HTTPStatus.CREATED
 
 
 @cors_preflight("GET, OPTIONS")
-@API.route("/task-outcome/<string:task_id>", methods=["GET", "OPTIONS"])
+@API.route("/task-outcome-configuration/<string:task_id>", methods=["GET", "OPTIONS"])
 @API.param("task_id", "Task ID")
 class TaskOutcomeByIdResource(Resource):
     """Resource to get task outcome configuration by task ID."""
@@ -83,5 +86,5 @@ class TaskOutcomeByIdResource(Resource):
     )
     def get(task_id: str):
         """Get task outcome configuration by task ID."""
-        response = TaskService().get_task_outcome(task_id)
+        response = TaskService().get_task_outcome_configuration(task_id)
         return response, HTTPStatus.OK

forms-flow-api/src/formsflow_api/schemas/__init__.py

@@ -47,5 +47,5 @@
 )
 from .process_history_logs import ProcessHistorySchema
 from .roles import RolesGroupsSchema
-from .tasks import TaskOutcomeSchema
+from .tasks import TaskOutcomeConfigurationSchema
 from .theme import ThemeCustomizationSchema

forms-flow-api/src/formsflow_api/schemas/tasks.py

@@ -5,16 +5,18 @@
 from .base_schema import AuditDateTimeSchema
 
 
-class TaskOutcomeSchema(AuditDateTimeSchema):
+class TaskOutcomeConfigurationSchema(AuditDateTimeSchema):
     """This class manages task outcome configuration schema."""
 
     class Meta:  # pylint: disable=too-few-public-methods
         """Exclude unknown fields in the deserialized output."""
 
         unknown = EXCLUDE
 
-    id = fields.Int()
+    id = fields.Int(dump_only=True)
     tenant = fields.Str(dump_only=True)
     task_id = fields.Str(data_key="taskId", required=True)
-    task_outcome = fields.List(fields.Dict(), data_key="taskOutcome", required=True)
+    task_name = fields.Str(data_key="taskName", required=True, allow_none=True)
+    task_transition_map = fields.Raw(data_key="taskTransitionMap", required=True)
+    transition_map_type = fields.Str(data_key="transitionMapType", required=True)
     created_by = fields.Str(data_key="createdBy", dump_only=True)

forms-flow-api/src/formsflow_api/services/tasks.py

@@ -1,56 +1,66 @@
 """This exposes tasks service."""
 
 from flask import current_app
+from formsflow_api_utils.exceptions import BusinessException
 from formsflow_api_utils.utils.user_context import UserContext, user_context
 
+from formsflow_api.constants import BusinessErrorCode
 from formsflow_api.models.tasks import TaskOutcomeConfiguration
-from formsflow_api.schemas.tasks import TaskOutcomeSchema
+from formsflow_api.schemas.tasks import TaskOutcomeConfigurationSchema
 
-task_outcome_schema = TaskOutcomeSchema()
+task_outcome_schema = TaskOutcomeConfigurationSchema()
 
 
 class TaskService:
     """This class manages task service."""
 
     @user_context
-    def create_task_outcome(
+    def create_task_outcome_configuration(
         self, request_data: dict, **kwargs
     ) -> TaskOutcomeConfiguration | None:
         """Create new task outcome."""
         current_app.logger.info("Creating task outcome configuration")
         user: UserContext = kwargs["user"]
         data = task_outcome_schema.load(request_data)
-        task_outcome = TaskOutcomeConfiguration.get_task_outcome_by_task_id(
-            data["task_id"], user.tenant_key
+        task_outcome_config = (
+            TaskOutcomeConfiguration.get_task_outcome_configuration_by_task_id(
+                data["task_id"], user.tenant_key
+            )
         )
-        if task_outcome:
+        if task_outcome_config:
             current_app.logger.info("Task outcome configuration already exists")
-            task_outcome.task_outcome = data["task_outcome"]
+            task_outcome_config.task_name = data["task_name"]
+            task_outcome_config.task_transition_map = data["task_transition_map"]
+            task_outcome_config.transition_map_type = data["transition_map_type"]
         else:
-            task_outcome = TaskOutcomeConfiguration(
+            task_outcome_config = TaskOutcomeConfiguration(
                 task_id=data["task_id"],
-                task_outcome=data["task_outcome"],
+                task_name=data["task_name"],
+                task_transition_map=data["task_transition_map"],
+                transition_map_type=data["transition_map_type"],
                 created_by=user.user_name,
                 tenant=user.tenant_key,
             )
-        task_outcome.save()
+        task_outcome_config.save()
         current_app.logger.info("Task outcome configuration created successfully")
-        response = task_outcome_schema.dump(task_outcome)
+        response = task_outcome_schema.dump(task_outcome_config)
         return response
 
     @user_context
-    def get_task_outcome(
+    def get_task_outcome_configuration(
         self, task_id: str, **kwargs
     ) -> TaskOutcomeConfiguration | None:
         """Get task outcome configuration by task ID."""
         current_app.logger.info(
             "Getting task outcome configuration for task ID {task_id}"
         )
         user: UserContext = kwargs["user"]
-        task_outcome = TaskOutcomeConfiguration.get_task_outcome_by_task_id(
-            task_id, user.tenant_key
+        task_outcome = (
+            TaskOutcomeConfiguration.get_task_outcome_configuration_by_task_id(
+                task_id, user.tenant_key
+            )
         )
         if task_outcome:
             response = task_outcome_schema.dump(task_outcome)
             return response
-        return None
+        raise BusinessException(BusinessErrorCode.TASK_OUTCOME_NOT_FOUND)