webapi: Feature/fwf 4504 task outcome configuration

CHANGELOG.md

@@ -36,6 +36,9 @@ Mark  items as `Added`, `Changed`, `Fixed`, `Modified`, `Removed`, `Untested Fea
 * Added variables(task_variables) as part of import and export.
 * Added Endpoint `/filter/filter-preference ` for saving user's filter preference data
 * Added new table called filter_preferences to handle filter preference of a user
+* Added new table TaskOutcomeConfiguration to store workflow transition rules
+* Added `/tasks/task-outcome-configuration` endpoint for task configuration storage
+* Added `/tasks/task-outcome-configuration/<task_id>` endpoint for task configuration lookup
 
 `Modified`
 

forms-flow-api/migrations/versions/5ecbbfc545ba_task_outcome_configuration_table.py

@@ -0,0 +1,42 @@
+"""Task_outcome_configuration_table
+
+Revision ID: 5ecbbfc545ba
+Revises: a5d9bbf7b5ac
+Create Date: 2025-05-07 11:29:25.237479
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '5ecbbfc545ba'
+down_revision = 'a5d9bbf7b5ac'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    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_name', sa.String(length=100), nullable=True, comment='Task name'),
+    sa.Column('task_transition_map', sa.JSON(), nullable=False, comment='Task transition configuration'),
+    sa.Column('transition_map_type', sa.String(length=100), nullable=False, server_default="select", comment='Task transition configuration 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')
+    )
+    op.create_index('idx_task_id_and_tenant', 'task_outcome_configuration', ['tenant', 'task_id'], unique=False)
+    op.create_index(op.f('ix_task_outcome_configuration_task_id'), 'task_outcome_configuration', ['task_id'], unique=True)
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f('ix_task_outcome_configuration_task_id'), table_name='task_outcome_configuration')
+    op.drop_index('idx_task_id_and_tenant', table_name='task_outcome_configuration')
+    op.drop_table('task_outcome_configuration')
+    # ### end Alembic commands ###

forms-flow-api/requirements/dev.txt

@@ -16,4 +16,5 @@ pytest-it
 pytest-mock
 pyflakes
 pylint
-lovely-pytest-docker
+lovely-pytest-docker
+snowballstemmer==2.2.0 # flake8-docstrings has incompatibility with snowballstemmer>=3

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

@@ -0,0 +1,43 @@
+"""This manages Task Data."""
+
+from __future__ import annotations
+
+from sqlalchemy import JSON, Index
+
+from .audit_mixin import ApplicationAuditDateTimeMixin
+from .base_model import BaseModel
+from .db import db
+
+
+class TaskOutcomeConfiguration(ApplicationAuditDateTimeMixin, BaseModel, db.Model):
+    """This class manages task outcome configurations."""
+
+    id = db.Column(db.Integer, primary_key=True)
+    task_id = db.Column(
+        db.String(100), nullable=False, comment="Task ID", unique=True, index=True
+    )
+    task_name = db.Column(db.String(100), nullable=True, comment="Task name")
+    task_transition_map = db.Column(
+        JSON, nullable=False, comment="Task transition configuration"
+    )
+    transition_map_type = db.Column(
+        db.String(100),
+        nullable=False,
+        default="select",
+        comment="Task transition configuration type",
+    )
+    created_by = db.Column(db.String(100), nullable=False, comment="Created by")
+    tenant = db.Column(db.String(100), nullable=True, comment="Tenant key")
+
+    __table_args__ = (Index("idx_task_id_and_tenant", "tenant", "task_id"),)
+
+    @classmethod
+    def get_task_outcome_configuration_by_task_id(
+        cls, task_id: str, tenant: str
+    ) -> TaskOutcomeConfiguration | None:
+        """Get task outcome configuration by task ID."""
+        query = cls.query.filter_by(task_id=task_id)
+        if tenant is not None:
+            query = query.filter_by(tenant=tenant)
+        task_outcome = query.first()
+        return task_outcome if task_outcome else None

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

@@ -30,6 +30,7 @@
 from formsflow_api.resources.process import API as PROCESS_API
 from formsflow_api.resources.public_endpoints import API as PUBLIC_API
 from formsflow_api.resources.roles import API as KEYCLOAK_ROLES_API
+from formsflow_api.resources.tasks import API as TASK_API
 from formsflow_api.resources.theme import API as THEME_CUSTOMIZATION_API
 from formsflow_api.resources.user import API as KEYCLOAK_USER_API
 
@@ -65,3 +66,4 @@
 API.add_namespace(INTEGRATION_API, path="/integrations")
 API.add_namespace(THEME_CUSTOMIZATION_API, path="/themes")
 API.add_namespace(IMPORT_API, path="/import")
+API.add_namespace(TASK_API, path="/tasks")

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

@@ -0,0 +1,136 @@
+"""API endpoints for managing tasks resource."""
+
+from http import HTTPStatus
+
+from flask import request
+from flask_restx import Namespace, Resource, fields
+from formsflow_api_utils.utils import (
+    auth,
+    cors_preflight,
+    profiletime,
+)
+
+from formsflow_api.services import TaskService
+
+API = Namespace("Tasks", description="Manages user tasks operations.")
+
+task_outcome_request = API.model(
+    "TaskOutcomeRequest",
+    {
+        "taskId": fields.String(description="Task ID", required=True),
+        "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 - accepts list, dict, string",
+            required=True,
+        ),
+    },
+)
+
+task_outcome_response = API.inherit(
+    "TaskOutcomeResponse",
+    task_outcome_request,
+    {
+        "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"),
+    },
+)
+
+
+@cors_preflight("POST, OPTIONS")
+@API.route("/task-outcome-configuration", methods=["POST", "OPTIONS"])
+class TaskOutcomeResource(Resource):
+    """Resource to create task outcome configuration."""
+
+    @staticmethod
+    @auth.require
+    @profiletime
+    @API.expect(task_outcome_request)
+    @API.doc(
+        responses={
+            201: ("CREATED:- Successful request.", task_outcome_response),
+            400: "BAD_REQUEST:- Invalid request.",
+            401: "UNAUTHORIZED:- Authorization header not provided or an invalid token passed.",
+        }
+    )
+    def post():
+        """Create task outcome configuration.
+
+        Accepts a JSON payload containing workflow configuration details.
+        The configuration includes transition rules and transition rules type for task outcomes.
+
+        Request Body:
+            dict: Required JSON payload with structure:
+                {
+                    "taskId": str,
+                    "taskName": str,
+                    "transitionMapType": str,     # "select", "radio", or "input"
+                    "taskTransitionMap": dict     # Outcome-to-step mappings supporting list, dict, string
+                }
+        Returns:
+            dict: Task outcome configuration with structure:
+                {
+                    "id": int,
+                    "taskId": str,
+                    "taskName": str,
+                    "tenant": str,
+                    "transitionMapType": str,  # "select", "radio", or "input"
+                    "taskTransitionMap": dict,  # mapping outcomes to subsequent workflow steps
+                    "created": str,
+                    "createdBy": str,
+                }
+        """
+        data = request.get_json()
+        if not data:
+            return {"message": "Invalid input"}, HTTPStatus.BAD_REQUEST
+        response = TaskService().create_task_outcome_configuration(data)
+        return response, HTTPStatus.CREATED
+
+
+@cors_preflight("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."""
+
+    @staticmethod
+    @auth.require
+    @profiletime
+    @API.doc(
+        responses={
+            200: ("OK:- Successful request.", task_outcome_response),
+            400: "BAD_REQUEST:- Invalid request.",
+            401: "UNAUTHORIZED:- Authorization header not provided or an invalid token passed.",
+        }
+    )
+    def get(task_id: str):
+        """Retrieves task outcome configuration by task ID.
+
+        Fetches the complete workflow configuration for a specified task, including
+        workflow routing rules (taskTransitionMap) and interface display preferences (transitionMapType),
+        supporting 'select', 'radio', or 'input'
+
+        Args:
+        task_id (str): Unique identifier of the task (required)
+
+        Returns:
+            dict: Task outcome configuration with structure:
+                {
+                    "id": int,
+                    "taskId": str,
+                    "taskName": str,
+                    "tenant": str,
+                    "transitionMapType": str,  # "select", "radio", or "input"
+                    "taskTransitionMap": dict,  # mapping outcomes to subsequent workflow steps
+                    "created": str,
+                    "createdBy": str,
+                }
+        """
+        response = TaskService().get_task_outcome_configuration(task_id)
+        return response, HTTPStatus.OK

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

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

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

@@ -0,0 +1,24 @@
+"""This manages Tasks Schema."""
+
+from marshmallow import EXCLUDE, fields
+
+from .base_schema import 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(dump_only=True)
+    tenant = fields.Str(dump_only=True)
+    task_id = fields.Str(data_key="taskId", required=True)
+    task_name = fields.Str(data_key="taskName", required=True, allow_none=True)
+    task_transition_map = fields.Raw(
+        data_key="taskTransitionMap", required=True
+    )  # Accepts list, dict, string
+    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/__init__.py

@@ -15,6 +15,7 @@
 from formsflow_api.services.form_process_mapper import FormProcessMapperService
 from formsflow_api.services.import_support import ImportService
 from formsflow_api.services.process import ProcessService
+from formsflow_api.services.tasks import TaskService
 from formsflow_api.services.theme import ThemeCustomizationService
 from formsflow_api.services.user import UserService
 
@@ -35,4 +36,5 @@
     "ThemeCustomizationService",
     "ImportService",
     "FilterPreferenceService",
+    "TaskService",
 ]

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

@@ -0,0 +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 TaskOutcomeConfigurationSchema
+
+task_outcome_schema = TaskOutcomeConfigurationSchema()
+
+
+class TaskService:
+    """This class manages task service."""
+
+    @user_context
+    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_config = (
+            TaskOutcomeConfiguration.get_task_outcome_configuration_by_task_id(
+                data["task_id"], user.tenant_key
+            )
+        )
+        if task_outcome_config:
+            current_app.logger.info("Task outcome configuration already exists")
+            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_config = TaskOutcomeConfiguration(
+                task_id=data["task_id"],
+                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_config.save()
+        current_app.logger.info("Task outcome configuration created successfully")
+        response = task_outcome_schema.dump(task_outcome_config)
+        return response
+
+    @user_context
+    def get_task_outcome_configuration(
+        self, task_id: str, **kwargs
+    ) -> TaskOutcomeConfiguration | None:
+        """Get task outcome configuration by task ID."""
+        current_app.logger.info(
+            f"Getting task outcome configuration for task ID {task_id}"
+        )
+        user: UserContext = kwargs["user"]
+        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
+        raise BusinessException(BusinessErrorCode.TASK_OUTCOME_NOT_FOUND)