"""This exposes form process mapper service.""" # pylint: disable=too-many-lines import json import re import xml.etree.ElementTree as ET from typing import List, Set, Tuple from flask import current_app from formsflow_api_utils.exceptions import BusinessException from formsflow_api_utils.services.external import FormioService from formsflow_api_utils.utils import CREATE_SUBMISSIONS, VIEW_SUBMISSIONS from formsflow_api_utils.utils.enums import FormProcessMapperStatus from formsflow_api_utils.utils.user_context import UserContext, user_context from formsflow_api.constants import ( BusinessErrorCode, default_flow_xml_data, default_task_variables, ) from formsflow_api.models import ( Application, Authorization, AuthType, FormHistory, FormProcessMapper, Process, ProcessStatus, ProcessType, ) from formsflow_api.schemas import ( FormProcessMapperRequestSchema, FormProcessMapperSchema, ProcessDataSchema, ) from formsflow_api.services.authorization import AuthorizationService from formsflow_api.services.external.bpm import BPMService from .filter import FilterService from .form_history_logs import FormHistoryService from .process import ProcessService class FormProcessMapperService: # pylint: disable=too-many-public-methods """This class manages form process mapper service.""" @staticmethod @user_context def get_all_forms( # pylint: disable=too-many-positional-arguments page_number: int, limit: int, search: list, sort_by: list, sort_order: list, form_type: str, is_active, is_designer: bool, active_forms: bool, include_submissions_count: bool, all_forms: bool = False, **kwargs, ): # pylint: disable=too-many-arguments, too-many-locals """Get all forms.""" user: UserContext = kwargs["user"] authorized_form_ids: Set[str] = [] current_app.logger.info(f"Listing forms for designer: {is_designer}") if active_forms: mappers, get_all_mappers_count = FormProcessMapper.find_all_active_forms( page_number=page_number, limit=limit, ) elif all_forms: # If all_forms is true, retrieve all forms regardless of their active status, # skipping authorization checks except for tenant-specific forms. mappers, get_all_mappers_count = FormProcessMapper.fetch_all_forms( page_number=page_number, limit=limit, ) else: form_ids = Authorization.find_all_resources_authorized( auth_type=AuthType.DESIGNER if is_designer else AuthType.FORM, roles=user.group_or_roles, user_name=user.user_name, tenant=user.tenant_key, include_created_by=is_designer, ) for forms in form_ids: authorized_form_ids.append(forms.resource_id) designer_filters = { "is_active": is_active, "form_type": form_type, } list_form_mappers = ( FormProcessMapper.find_all_forms if is_designer else Application.find_all_active_by_formid ) # Submissions count should return for user with create_submissions or view_submissions permission fetch_submissions_count = include_submissions_count and any( perm in user.roles for perm in [CREATE_SUBMISSIONS, VIEW_SUBMISSIONS] ) mappers, get_all_mappers_count = list_form_mappers( page_number=page_number, limit=limit, search=search, sort_by=sort_by, sort_order=sort_order, form_ids=authorized_form_ids, **( designer_filters if is_designer else {"fetch_submissions_count": fetch_submissions_count} ), ) mapper_schema = FormProcessMapperSchema() mappers_response = mapper_schema.dump(mappers, many=True) return ( mappers_response, get_all_mappers_count, ) @staticmethod def sort_results(data: List, sort_order: str, sort_by: str): """Sort results.""" reverse = ( "desc" in sort_order ) # Determine if sorting should be in descending order return sorted(data, key=lambda k: k.get(sort_by, 0), reverse=reverse) @staticmethod def get_mapper_count(form_name=None): """Get form process mapper count.""" if form_name: return FormProcessMapper.find_count_form_name(form_name) return FormProcessMapper.find_all_count() @staticmethod @user_context def get_mapper(form_process_mapper_id: int, **kwargs): """Get form process mapper.""" user: UserContext = kwargs["user"] tenant_key = user.tenant_key mapper = FormProcessMapper.find_form_by_id_active_status( form_process_mapper_id=form_process_mapper_id ) if mapper: if tenant_key is not None and mapper.tenant != tenant_key: raise PermissionError("Tenant authentication failed.") mapper_schema = FormProcessMapperSchema() return mapper_schema.dump(mapper) raise BusinessException(BusinessErrorCode.INVALID_FORM_PROCESS_MAPPER_ID) @staticmethod def get_form_version(mapper): """Get form versions.""" version_data = FormHistory.get_latest_version(mapper.parent_form_id) major_version, minor_version = 1, 0 if version_data: major_version = version_data.major_version minor_version = version_data.minor_version return major_version, minor_version @staticmethod @user_context def get_mapper_by_formid(form_id: str, **kwargs): """Get form process mapper.""" user: UserContext = kwargs["user"] tenant_key = user.tenant_key mapper = FormProcessMapper.find_form_by_form_id(form_id=form_id) if mapper: if tenant_key is not None and mapper.tenant != tenant_key: raise PermissionError("Tenant authentication failed.") mapper_schema = FormProcessMapperSchema() response = mapper_schema.dump(mapper) # Include form versions major_version, minor_version = FormProcessMapperService.get_form_version( mapper ) response["majorVersion"] = major_version response["minorVersion"] = minor_version if response.get("deleted") is False: return response raise BusinessException(BusinessErrorCode.INVALID_FORM_PROCESS_MAPPER_ID) @staticmethod @user_context def create_mapper(data, **kwargs): """Create new mapper between form and process.""" user: UserContext = kwargs["user"] data["created_by"] = user.user_name data["tenant"] = user.tenant_key data["process_tenant"] = user.tenant_key return FormProcessMapper.create_from_dict(data) @staticmethod def _update_process_tenant(data, user): # For multi tenant environment find if the process is deployed for a tenant. if current_app.config.get("MULTI_TENANCY_ENABLED") and ( process_key := data.get("process_key", None) ): current_app.logger.info("Finding Tenant ID for process %s ", process_key) data["process_tenant"] = BPMService.get_process_details_by_key( process_key, user.bearer_token ).get("tenantId", None) @staticmethod @user_context def update_mapper(form_process_mapper_id, data, **kwargs): """Update form process mapper.""" user: UserContext = kwargs["user"] tenant_key = user.tenant_key data["modified_by"] = user.user_name mapper = FormProcessMapper.find_form_by_id( form_process_mapper_id=form_process_mapper_id ) if not data.get("comments"): data["comments"] = None if mapper: if tenant_key is not None and mapper.tenant != tenant_key: raise BusinessException(BusinessErrorCode.PERMISSION_DENIED) mapper.update(data) return mapper raise BusinessException(BusinessErrorCode.INVALID_FORM_PROCESS_MAPPER_ID) @staticmethod @user_context def mark_inactive_and_delete(form_process_mapper_id: int, **kwargs) -> None: """Mark form process mapper as inactive and deleted.""" user: UserContext = kwargs["user"] tenant_key = user.tenant_key application = FormProcessMapper.find_form_by_id( form_process_mapper_id=form_process_mapper_id ) if application: if tenant_key is not None and application.tenant != tenant_key: raise PermissionError("Tenant authentication failed.") count = Application.get_total_application_corresponding_to_mapper_id( form_process_mapper_id ) if count > 0: raise BusinessException(BusinessErrorCode.RESTRICT_FORM_DELETE) application.mark_inactive() # fetching all draft application application and delete it draft_applications = Application.get_draft_by_parent_form_id( parent_form_id=application.parent_form_id ) if draft_applications: for draft in draft_applications: draft.delete() else: raise BusinessException(BusinessErrorCode.INVALID_FORM_PROCESS_MAPPER_ID) @staticmethod def mark_unpublished(form_process_mapper_id): """Mark form process mapper as inactive.""" mapper = FormProcessMapper.find_form_by_id( form_process_mapper_id=form_process_mapper_id ) if mapper: mapper.mark_unpublished() return raise BusinessException(BusinessErrorCode.INVALID_FORM_PROCESS_MAPPER_ID) @staticmethod def get_mapper_by_formid_and_version(form_id: int, version: int): """Returns a serialized form process mapper given a form_id and version.""" mapper = FormProcessMapper.find_mapper_by_form_id_and_version(form_id, version) if mapper: mapper_schema = FormProcessMapperSchema() return mapper_schema.dump(mapper) return None @staticmethod def unpublish_previous_mapper(mapper_data: dict) -> None: """ This method unpublishes the previous version of the form process mapper. : mapper_data: serialized create mapper payload : Should be called with create_mapper method """ form_id = mapper_data.get("previous_form_id") or mapper_data.get("form_id") version = mapper_data.get("version") if version is None or form_id is None: return version = int(version) - 1 previous_mapper = FormProcessMapperService.get_mapper_by_formid_and_version( form_id, version ) previous_status = previous_mapper.get("status") if previous_mapper and previous_status == FormProcessMapperStatus.ACTIVE.value: previous_mapper_id = previous_mapper.get("id") FormProcessMapperService.mark_unpublished(previous_mapper_id) @staticmethod @user_context def check_tenant_authorization(mapper_id: int, **kwargs) -> int: """Check if tenant has permission to access the resource.""" user: UserContext = kwargs["user"] tenant_key = user.tenant_key if tenant_key is None: return 0 mapper = FormProcessMapper.find_form_by_id(form_process_mapper_id=mapper_id) if mapper is not None and mapper.tenant != tenant_key: raise BusinessException(BusinessErrorCode.PERMISSION_DENIED) return 0 @staticmethod @user_context def check_tenant_authorization_by_formid( form_id: int, mapper_data=None, **kwargs ) -> int: """Check if tenant has permission to access the resource.""" user: UserContext = kwargs["user"] tenant_key = user.tenant_key if tenant_key is None: return # If mapper data is provided as an argument, there's no need to fetch it from the database mapper = ( mapper_data if mapper_data else FormProcessMapper.find_form_by_form_id(form_id=form_id) ) if mapper is not None and mapper.tenant != tenant_key: raise BusinessException(BusinessErrorCode.PERMISSION_DENIED) return @staticmethod def validate_process_and_update_mapper(name, mapper): """Validate process name/key exists, if exists update name & update mapper.""" current_app.logger.info(f"Validating process key already exists. {name}") process = Process.find_process_by_name_key(name=name, process_key=name) if process: # Since the process key/name already exists create updated process key by appending mapper Id # Update mapper with updated value updated_process_name = f"{name}_{mapper.id}" mapper.process_key = updated_process_name mapper.process_name = updated_process_name mapper.save() return updated_process_name return None @staticmethod def mapper_create(mapper_json): """Service to handle mapper create.""" current_app.logger.debug("Creating mapper..") mapper_json["taskVariables"] = json.dumps( mapper_json.get("taskVariables") or [] ) mapper_schema = FormProcessMapperSchema() dict_data = mapper_schema.load(mapper_json) mapper = FormProcessMapperService.create_mapper(dict_data) FormProcessMapperService.unpublish_previous_mapper(dict_data) return mapper @staticmethod def form_design_update(data, form_id): """Service to handle form design update.""" mapper = FormProcessMapper.find_form_by_form_id(form_id=form_id) FormProcessMapperService.check_tenant_authorization_by_formid( form_id=form_id, mapper_data=mapper ) formio_service = FormioService() form_io_token = formio_service.get_formio_access_token() response = formio_service.update_form(form_id, data, form_io_token) # if user selected to continue with minor version after unpublish if mapper.prompt_new_version: mapper.update({"prompt_new_version": False}) FormHistoryService.create_form_log_with_clone(data=data) return response @classmethod @user_context def create_default_process(cls, process_name, status=ProcessStatus.DRAFT, **kwargs): """Create process with default workflow.""" user: UserContext = kwargs["user"] process_dict = { "name": process_name, "process_key": process_name, "parent_process_key": process_name, "process_type": ProcessType.BPMN, "status": status, "process_data": default_flow_xml_data(process_name).encode("utf-8"), "tenant": user.tenant_key, "major_version": 1, "minor_version": 0, "created_by": user.user_name, } process = Process.create_from_dict(process_dict) return process @classmethod def create_formio_form( cls, data, ): """Service to handle form create in formio.""" current_app.logger.info("Creating form in formio..") # Initialize formio service and get formio token to create the form formio_service = FormioService() form_io_token = formio_service.get_formio_access_token() # creating form and get response from formio response = formio_service.create_form(data, form_io_token) return response @classmethod def create_authorization_for_form( cls, parent_form_id, is_designer, user, authorization_data=None ): """ Creates or updates authorization settings for a form resource. This method either updates the provided authorization data with the correct resource ID or generates default authorization data for the given form. It then calls the AuthorizationService to create or update the resource authorization accordingly. Args: parent_form_id (str): The unique identifier of the parent form. is_designer (bool): Flag indicating if the user is a designer. user (User): The user object containing user information. authorization_data (dict, optional): Existing authorization data to update. If None, default authorization data will be created. Returns: dict: The authorization data used to create or update the resource authorization. """ if authorization_data: # Incase of combine save of form and authorization settings the resourceId is unknown # So update with parent form id for section in authorization_data: if authorization_data[section].get("resourceId") is None: authorization_data[section]["resourceId"] = parent_form_id else: # create default data for authorization of the resource authorization_data = { "application": { "resourceId": parent_form_id, "resourceDetails": {"submitter": True}, "roles": [], "userName": None, }, "designer": { "resourceId": parent_form_id, "resourceDetails": {}, "roles": [], "userName": user.user_name, }, "form": { "resourceId": parent_form_id, "resourceDetails": {}, "roles": [], }, } current_app.logger.debug( "Creating default data for authorization of the resource.." ) AuthorizationService.create_or_update_resource_authorization( authorization_data, is_designer=is_designer ) return authorization_data @classmethod def create_process(cls, process_data, process_type, process_name): """ Creates a process based on the provided process data and type, or creates a default process if not provided. Args: process_data (dict): The data required to create the process. If None, a default process is created. process_type (str): The type of the process to be created. Required if process_data is provided. process_name (str): The name of the process to be created. Returns: dict: The response from the process creation. """ if process_data and process_type: # Incase of duplicate form we get process data from payload process_response = ProcessService.create_process( process_data, process_type, process_name, process_name ) else: # create entry in process with default flow. process_response = FormProcessMapperService.create_default_process( process_name ) return process_response @staticmethod @user_context def create_form( data, is_designer, combine_save=False, **kwargs ): # pylint:disable=too-many-locals """Service to handle form create.""" current_app.logger.info("Creating form..") user: UserContext = kwargs["user"] # create the form in formo response = FormProcessMapperService.create_formio_form(data) form_id = response.get("_id") parent_form_id = data.get("parentFormId", form_id) # is_new_form=True if creating a new form, False if creating a new version is_new_form = parent_form_id == form_id process_key = None anonymous = False if not data.get("anonymous") else data.get("anonymous") description = data.get("description", "") task_variable = ( [*default_task_variables] if not data.get("taskVariables") else data.get("taskVariables") ) is_migrated = True current_app.logger.info(f"Creating new form {is_new_form}") # If creating new version for a existing form, fetch process key, name from mapper if not is_new_form: current_app.logger.debug("Fetching details from mapper") mapper = FormProcessMapper.get_latest_by_parent_form_id(parent_form_id) process_name = mapper.process_name process_key = mapper.process_key anonymous = mapper.is_anonymous description = mapper.description task_variable = json.loads(mapper.task_variable) is_migrated = mapper.is_migrated else: # if new form, form name is kept as process_name & process key process_name = response.get("name") # process key/Id doesn't support numbers & special characters at start # special characters anywhere so clean them before setting as process key process_name = ProcessService.clean_form_name(process_name) mapper_data = { "formId": form_id, "formName": response.get("title"), "description": description, "formType": response.get("type"), "processKey": process_name, "processName": process_key if process_key else process_name, "formTypeChanged": True, "parentFormId": parent_form_id, "titleChanged": True, "formRevisionNumber": "V1", "status": FormProcessMapperStatus.INACTIVE.value, "anonymous": anonymous, "taskVariables": task_variable, "isMigrated": is_migrated, } mapper = FormProcessMapperService.mapper_create(mapper_data) current_app.logger.debug("Creating form log with clone..") FormHistoryService.create_form_log_with_clone( data={ **response, "parentFormId": parent_form_id, "newVersion": True, "componentChanged": True, } ) if is_new_form: authorization_data = FormProcessMapperService.create_authorization_for_form( parent_form_id, is_designer, user, data.get("authorizations") ) # validate process key already exists, if exists append mapper id to process_key. updated_process_name = ( FormProcessMapperService.validate_process_and_update_mapper( process_name, mapper ) ) process_name = ( updated_process_name if updated_process_name else process_name ) process_data = data.get("processData") process_type = data.get("processType") process_data_response = FormProcessMapperService.create_process( process_data, process_type, process_name ) if combine_save: mapper_response = FormProcessMapperSchema().dump(mapper) process_response = ProcessDataSchema().dump(process_data_response) if task_variables := mapper_response.get("taskVariables"): mapper_response["taskVariables"] = json.loads(task_variables) response = { "formData": response, "authorizations": authorization_data, "mapper": mapper_response, "process": process_response, } return response def _remove_tenant_key(self, form_json, tenant_key): """Remove tenant key from path & name.""" tenant_prefix = f"{tenant_key}-" form_path = form_json.get("path", "") form_name = form_json.get("name", "") current_app.logger.info( f"Removing tenant key from path: {form_path} & name: {form_name}" ) if form_path.startswith(tenant_prefix): form_json["path"] = form_path[len(tenant_prefix) :] if form_name.startswith(tenant_prefix): form_json["name"] = form_name[len(tenant_prefix) :] return form_json def _sanitize_form_json(self, form_json, tenant_key): """Clean form JSON data for export.""" keys_to_remove = [ "_id", "machineName", "access", "submissionAccess", "parentFormId", "owner", "tenantKey", ] for key in keys_to_remove: form_json.pop(key, None) # Remove 'tenantkey-' from 'path' and 'name' if current_app.config.get("MULTI_TENANCY_ENABLED"): form_json = self._remove_tenant_key(form_json, tenant_key) return form_json def _get_form( # pylint: disable=too-many-arguments, too-many-positional-arguments self, title_or_path: str, scope_type: str, form_id: str = None, description: str = None, tenant_key: str = None, anonymous: bool = False, task_variable=None, ) -> dict: """Get form details.""" try: current_app.logger.info(f"Fetching form : {title_or_path}") formio_service = FormioService() form_io_token = formio_service.get_formio_access_token() if form_id: form_json = formio_service.get_form_by_id(form_id, form_io_token) else: form_json = formio_service.get_form_by_path( title_or_path, form_io_token ) if not form_json: raise BusinessException(BusinessErrorCode.INVALID_FORM_ID) # In a (sub form)connected form, the workflow provides the form path, # and the title is obtained from the form JSON title_or_path = ( form_json.get("title", "") if scope_type == "sub" else title_or_path ) form_json = self._sanitize_form_json(form_json, tenant_key) return { "formTitle": title_or_path, "formDescription": description, "anonymous": anonymous or False, "type": scope_type, "taskVariable": task_variable, "content": form_json, } except Exception as e: current_app.logger.error(e) raise BusinessException(BusinessErrorCode.FORM_ID_NOT_FOUND) from e def _get_workflow( self, process_key: str, process_name: str, scope_type: str ) -> dict: """Get workflow details.""" current_app.logger.info(f"Fetching Process : {process_key}") process = Process.get_latest_version_by_key(process_key) if process: process_data = process.process_data.decode("utf-8") process_type = process.process_type.value content = ( json.loads(process_data) if process_type == "LOWCODE" else process_data ) return { "processKey": process_key, "processName": process_name, "processType": process_type, "type": scope_type, "content": content, } raise BusinessException(BusinessErrorCode.PROCESS_DEF_NOT_FOUND) def _get_dmn(self, dmn_key: str, scope_type: str, user: UserContext) -> dict: """Get DMN details.""" try: current_app.logger.info(f"Fetching xml for DMN: {dmn_key}") dmn_tenant = None if current_app.config.get("MULTI_TENANCY_ENABLED"): url_path = ( f"?latestVersion=true&includeDecisionDefinitionsWithoutTenantId=true" f"&key={dmn_key}&tenantIdIn={user.tenant_key}" ) dmn = BPMService.get_decision(user.bearer_token, url_path) if dmn: dmn_tenant = dmn[0].get("tenantId") current_app.logger.info( f"Found tenant ID: {dmn_tenant} for DMN: {dmn_key}" ) dmn_xml = BPMService.decision_definition_xml( dmn_key, user.bearer_token, dmn_tenant ).get("dmnXml") return { "key": dmn_key, "type": scope_type, "content": dmn_xml, } except Exception as e: current_app.logger.error(e) raise BusinessException(BusinessErrorCode.DECISION_DEF_NOT_FOUND) from e def _get_authorizations(self, resource_id: str, user) -> dict: """Get authorization details.""" auth_details = Authorization.find_auth_list_by_id(resource_id, user.tenant_key) auth_detail = {} for auth in auth_details: auth_detail[auth.auth_type.value] = { "resourceId": auth.resource_id, "resourceDetails": auth.resource_details, "roles": auth.roles, "userName": None, } return auth_detail def _parse_xml( # pylint:disable=too-many-locals self, bpmn_xml: str, user: UserContext ) -> Tuple[List[str], List[str], List[dict]]: """Parse the XML string.""" current_app.logger.info("Parsing XML...") root = ET.fromstring(bpmn_xml) namespaces = { "bpmn": "http://www.omg.org/spec/BPMN/20100524/MODEL", "camunda": "http://camunda.org/schema/1.0/bpmn", } form_names = [] dmn_names = [] workflows = [] # Find all 'camunda:taskListener' with class="FormConnectorListener" current_app.logger.info("Search for task with form connector...") form_connector_tasks = root.findall( ".//camunda:taskListener" "[@class='org.camunda.bpm.extension.hooks.listeners.task.FormConnectorListener']" "/../camunda:properties/camunda:property", namespaces, ) for task in form_connector_tasks: if task.get("name") == "formName": form_names.append(task.get("value")) current_app.logger.info(f"Forms found: {form_names}") # Find DMNs current_app.logger.info("Search for task with DMN...") dmn_tasks = root.findall( ".//bpmn:businessRuleTask[@camunda:decisionRef]", namespaces ) for task in dmn_tasks: decision_ref = task.attrib.get( "{http://camunda.org/schema/1.0/bpmn}decisionRef" ) dmn_names.append(decision_ref) current_app.logger.info( f"Task ID: {task.attrib.get('id')}, DMN: {decision_ref}" ) # Find subprocesses current_app.logger.info("Search for subprocess...") sub_processes = root.findall(".//bpmn:callActivity[@calledElement]", namespaces) for subprocess in sub_processes: subprocess_name = subprocess.attrib.get("calledElement") current_app.logger.info(f"Subprocess: {subprocess_name}") # Here subprocess_name will be the process key # Since we didn't get process name, we will use the subprocess name as process name sub_workflow = self._get_workflow(subprocess_name, subprocess_name, "sub") workflows.append(sub_workflow) sub_form_names, sub_dmn_names, sub_workflows = self._parse_xml( sub_workflow["content"], user ) form_names.extend(sub_form_names) dmn_names.extend(sub_dmn_names) workflows.extend(sub_workflows) return form_names, dmn_names, workflows @user_context def export( # pylint:disable=too-many-locals self, mapper_id: int, **kwargs ) -> dict: """Export form & workflow.""" current_app.logger.info(f"Exporting form process mapper: {mapper_id}") mapper = FormProcessMapper.find_form_by_id(form_process_mapper_id=mapper_id) user: UserContext = kwargs["user"] tenant_key = user.tenant_key if mapper: if tenant_key is not None and mapper.tenant != tenant_key: raise PermissionError(BusinessErrorCode.PERMISSION_DENIED) forms = [] workflows = [] rules = [] authorizations = [] # Capture main form & workflow forms.append( self._get_form( mapper.form_name, "main", mapper.form_id, mapper.description, tenant_key, mapper.is_anonymous, mapper.task_variable, ) ) workflow = self._get_workflow( mapper.process_key, mapper.process_name, "main" ) workflows.append(workflow) authorizations.append(self._get_authorizations(mapper.parent_form_id, user)) # Parse bpm xml to get subforms & workflows # The following lines are currently commented out but may be needed for future use. # forms_names, dmns, sub_workflows = self._parse_xml( # workflow["content"], user # ) # for form in set(forms_names): # forms.append( # self._get_form(form, "sub", form_id=None, description=None, tenant_key=tenant_key) # ) # for dmn in set(dmns): # rules.append(self._get_dmn(dmn, "sub", user)) # workflows.extend(sub_workflows) return { "forms": forms, "workflows": workflows, "rules": rules, "authorizations": authorizations, } raise BusinessException(BusinessErrorCode.INVALID_FORM_PROCESS_MAPPER_ID) @classmethod def is_valid_field(cls, field: str, pattern: str) -> bool: """Checks if the given field matches the provided regex pattern.""" return bool(re.fullmatch(pattern, field)) @classmethod def validate_title_name_path(cls, title: str, path: str, name: str): """Validates the title, path, and name fields.""" title_pattern = r"(?=.*[A-Za-z])^[A-Za-z0-9 ]+(-{1,}[A-Za-z0-9 ]+)*$" path_name = r"(?=.*[A-Za-z])^[A-Za-z0-9]+(-{1,}[A-Za-z0-9]+)*$" invalid_fields = [] error_messages = { "title": "Title: Only contain alphanumeric characters, hyphens(not at the start or end), spaces," "and must include at least one letter.", "path": "Path: Only contain alphanumeric characters, hyphens(not at the start or end), no spaces," "and must include at least one letter.", "name": "Name: Only contain alphanumeric characters, hyphens(not at the start or end), no spaces," "and must include at least one letter.", } # Validate title if title and not cls.is_valid_field(title, title_pattern): invalid_fields.append("title") # Validate path and name for field_name, field_value in (("path", path), ("name", name)): if field_value and not cls.is_valid_field(field_value, path_name): invalid_fields.append(field_name) # Determine overall validity is_valid = len(invalid_fields) == 0 if not is_valid: # Generate detailed validation error message error_message = ",\n ".join( error_messages[field] for field in invalid_fields ) raise BusinessException( BusinessErrorCode.FORM_VALIDATION_FAILED, detail_message=error_message, include_details=True, ) @classmethod def validate_form_title(cls, title, exclude_id=None): """Validate form tile in the form_process_mapper table.""" # Exclude the current mapper from the query current_app.logger.info( f"Validation for form title...{title}..with exclude id-{exclude_id}" ) mappers = FormProcessMapper.find_forms_by_title(title, exclude_id=exclude_id) if mappers: current_app.logger.debug(f"Other mappers matching the title- {mappers}") raise BusinessException(BusinessErrorCode.FORM_EXISTS) return True @staticmethod def validate_query_parameters(title, name, path): """Check if at least one query parameter is provided.""" if not (title or name or path): raise BusinessException(BusinessErrorCode.INVALID_FORM_VALIDATION_INPUT) @staticmethod def validate_path(path): """Validate path with formio resevered keywords.""" current_app.logger.debug(f"Validate path for reseverd keyword:{path}") # Keywords that are invalid as standalone input restricted_keywords = { "exists", "export", "role", "current", "logout", "import", "form", "access", "token", "recaptcha", } # Forbidden end keywords forbidden_end_keywords = {"submission", "action"} if ( path in restricted_keywords or path and any(path.endswith(keyword) for keyword in forbidden_end_keywords) ): raise BusinessException(BusinessErrorCode.INVALID_PATH) return True @staticmethod @user_context def validate_form_name_path_title(request, **kwargs): """Validate a form name by calling the external validation API.""" # Retrieve the parameters from the query string title = request.args.get("title") name = request.args.get("name") path = request.args.get("path") form_id = request.args.get("id") parent_form_id = request.args.get("parentFormId") current_app.logger.info( f"Title:{title}, Name:{name}, Path:{path}, form_id:{form_id}, parent_form_id: {parent_form_id}" ) FormProcessMapperService.validate_query_parameters(title, name, path) if title and len(title) > 200: raise BusinessException(BusinessErrorCode.INVALID_FORM_TITLE_LENGTH) FormProcessMapperService.validate_title_name_path(title, path, name) # In case of new form creation, title alone passed form UI # Trim space & validate path if not parent_form_id and title: path = title.replace(" ", "") if current_app.config.get("MULTI_TENANCY_ENABLED"): user: UserContext = kwargs["user"] tenant_key = user.tenant_key name = f"{tenant_key}-{name}" path = f"{tenant_key}-{path}" # Validate path has reserved keywords FormProcessMapperService.validate_path(path) # Validate title exists validation on mapper & path, name in formio. if title: FormProcessMapperService.validate_form_title(title, parent_form_id) # Validate path, name exits in formio. if path or name: query_params = f"name={name}&path={path}&select=title,path,name" # Initialize the FormioService and get the access token formio_service = FormioService() form_io_token = formio_service.get_formio_access_token() validation_response = formio_service.get_form_search( query_params, form_io_token ) # Check if the validation response has any results if validation_response: # Check if the form ID matches if ( form_id and len(validation_response) == 1 and validation_response[0].get("_id") == form_id ): return {} # If there are results but no matching ID, the form name is still considered invalid raise BusinessException(BusinessErrorCode.FORM_EXISTS) # If no results, the form name is valid return {} def validate_mapper(self, mapper_id, tenant_key): """Validate mapper by mapper Id.""" mapper = FormProcessMapper.find_form_by_id(form_process_mapper_id=mapper_id) if not mapper: raise BusinessException(BusinessErrorCode.INVALID_FORM_PROCESS_MAPPER_ID) # Check tenant authentication if tenant_key and mapper.tenant != tenant_key: raise PermissionError(BusinessErrorCode.PERMISSION_DENIED) # Check the mapper_id provided is the latest mapper for the specific parent_form_id. latest = FormProcessMapper.get_latest_by_parent_form_id(mapper.parent_form_id) if latest and mapper.id != latest.id: raise BusinessException(BusinessErrorCode.MAPPER_NOT_LATEST_VERSION) return mapper def capture_form_history(self, mapper, data, user_name): """Capture form history.""" major_version, minor_version = 1, 0 latest_form_history = FormHistory.get_latest_version(mapper.parent_form_id) if latest_form_history: major_version, minor_version = ( latest_form_history.major_version, latest_form_history.minor_version, ) FormHistory( created_by=user_name, parent_form_id=mapper.parent_form_id, form_id=mapper.form_id, change_log=data, status=True, major_version=major_version, minor_version=minor_version, ).save() @user_context def publish(self, mapper_id, **kwargs): """Publish by mapper_id.""" user: UserContext = kwargs["user"] tenant_key = user.tenant_key token = user.bearer_token user_name = user.user_name mapper = self.validate_mapper(mapper_id, tenant_key) process_name = mapper.process_key # Fetch process data from process table process = Process.get_latest_version_by_key(process_name) process_data, process_type = ( (process.process_data, process.process_type) if process else (None, None) ) # Deploy process ProcessService.deploy_process( process_name, process_data, tenant_key, token, process_type ) if not process: # create entry in process with default flow with status "PUBLISHED". FormProcessMapperService.create_default_process( process_name, status=ProcessStatus.PUBLISHED ) else: # Update process status ProcessService.update_process_status(process, ProcessStatus.PUBLISHED, user) # Capture publish(active) status in form history table. self.capture_form_history(mapper, {"status": "active"}, user_name) # Update status in mapper table mapper.update( { "status": str(FormProcessMapperStatus.ACTIVE.value), "prompt_new_version": False, } ) return {} @user_context def unpublish(self, mapper_id: int, **kwargs): """Publish by mapper_id.""" user: UserContext = kwargs["user"] user_name = user.user_name tenant_key = user.tenant_key mapper = self.validate_mapper(mapper_id, tenant_key) # Capture unpublish status in form history table. self.capture_form_history(mapper, {"status": "inactive"}, user_name) # Update status(inactive) in mapper table mapper.update( { "status": str(FormProcessMapperStatus.INACTIVE.value), "prompt_new_version": True, } ) # Update process status to Draft process = Process.get_latest_version_by_key(mapper.process_key) if process: ProcessService.update_process_status(process, ProcessStatus.DRAFT, user) return {} @user_context def get_form_data(self, form_id: str, auth_type: str, is_designer: bool, **kwargs): """Get form data by form_id.""" user: UserContext = kwargs["user"] tenant_key = user.tenant_key auth_type = auth_type.upper() if auth_type else AuthType.FORM.value mapper_data = FormProcessMapper.find_form_by_form_id(form_id) # check the mapper exists and is accessible if not mapper_data: raise BusinessException(BusinessErrorCode.FORM_NOT_FOUND) if auth_type == AuthType.FORM.value: # check the form is published if mapper_data.status == FormProcessMapperStatus.INACTIVE.value: raise BusinessException(BusinessErrorCode.FORM_NOT_PUBLISHED) # check the user is not anonymous then the tenant must match if not mapper_data.is_anonymous and mapper_data.tenant != tenant_key: raise PermissionError(BusinessErrorCode.PERMISSION_DENIED) auth_service = AuthorizationService() auth_data = None if auth_type == AuthType.APPLICATION.value: auth_data = auth_service.get_application_resource_by_id( auth_type=auth_type, resource_id=mapper_data.parent_form_id, form_id=form_id, user=user, ) else: auth_data = auth_service.get_resource_by_id( auth_type=auth_type, resource_id=mapper_data.parent_form_id, is_designer=( is_designer if auth_type == AuthType.DESIGNER.value else False ), user=user, ) if not auth_data: raise BusinessException(BusinessErrorCode.PERMISSION_DENIED) formio_service = FormioService() form_io_token = formio_service.get_formio_access_token() form_json = formio_service.get_form_by_id(form_id, form_io_token) return form_json @classmethod def create_form_with_process(cls, request_data, is_designer): """Create a form with process and authorization in a single operation. Args: request_data (dict): Contains form data, process data and authorization { formData (dict): Form configuration data processData (str): BPMN XML string or JSON for low code process processType (str): Type of process (BPMN or LOWCODE) authorizations (dict): Authorization configuration taskVariables (list): List of task variables } is_designer (bool): Whether the request is from a designer Returns: dict: Created form and associated process details """ data = request_data.get("formData") if not data: raise BusinessException(BusinessErrorCode.FORM_PAYLOAD_MISSING) process_data = request_data.get("processData") process_type = request_data.get("processType") authorizations = request_data.get("authorizations") task_variables = request_data.get("taskVariables") # Add task variables to form data if task_variables: data["taskVariables"] = task_variables if process_data and process_type: data["processData"] = process_data data["processType"] = process_type if authorizations: data["authorizations"] = authorizations # Create form and associated process response = FormProcessMapperService.create_form( data, is_designer, combine_save=True ) return response def handle_form_data(self, form_data, **kwargs): # pylint:disable=unused-argument """Handler function for form data updates.""" current_app.logger.debug("Updating form design..") form_id = form_data.get("_id") if not form_id: raise BusinessException(BusinessErrorCode.INVALID_INPUT) return FormProcessMapperService.form_design_update(form_data, form_id) def handle_mapper_data( self, mapper_data, mapper_id=None, **kwargs ): # pylint:disable=unused-argument """Handler function for mapper data updates.""" if mapper_id is None: raise BusinessException(BusinessErrorCode.INVALID_INPUT) current_app.logger.debug("Updating mapper details..") task_variable = mapper_data.get("taskVariables", []) # If task variables are present, update filter variables and serialize them if "taskVariables" in mapper_data: FilterService.update_filter_variables( task_variable, mapper_data.get("formId") ) mapper_data["taskVariables"] = json.dumps(task_variable) # Load the mapper data into the schema dict_data = FormProcessMapperRequestSchema().load(mapper_data) mapper = FormProcessMapperService.update_mapper(mapper_id, dict_data) # Dump the updated mapper data into the response schema mapper_response = FormProcessMapperSchema().dump(mapper) if task_variables := mapper_response.get("taskVariables"): mapper_response["taskVariables"] = json.loads(task_variables) # Get major and minor version of the from from form history major_version, minor_version = FormProcessMapperService.get_form_version(mapper) mapper_response["majorVersion"] = major_version mapper_response["minorVersion"] = minor_version return mapper_response def handle_authorization_data( self, authorization_data, is_designer=False, **kwargs ): # pylint:disable=unused-argument """Handler function for authorization updates.""" current_app.logger.debug("Updating authorization details..") AuthorizationService.create_or_update_resource_authorization( authorization_data, is_designer=is_designer ) return authorization_data def handle_process_data( self, process_data, **kwargs ): # pylint:disable=unused-argument """Handler function for process updates.""" current_app.logger.debug("Updating process details..") process_id = process_data.get("id") if not process_id: raise BusinessException(BusinessErrorCode.INVALID_INPUT) process_payload = process_data.get("processData") process_type = process_data.get("processType") return ProcessService.update_process(process_id, process_payload, process_type) def validate_request_data(self, request_data, expected_keys): """Check if at least one expected key exists and is not None.""" has_valid_data = any(request_data.get(key) is not None for key in expected_keys) if not has_valid_data: raise BusinessException(BusinessErrorCode.MISSING_REQUIRED_KEYS) return True def update_form_process(self, request_data, mapper_id, is_designer): """Update form, process, authorizations and mapper details in one call. Args: request_data (dict): Combined update payload that may include any of: - formData (dict): Formio form JSON; must include `_id` when provided. - mapper (dict): Mapper fields to update (e.g., formName, status, taskVariables, etc.). If `taskVariables` is present, it will be persisted and filter variables will be updated accordingly. - authorizations (dict): Authorization configuration; created/updated for the associated resource when provided. - process (dict): { id, processData, processType } used to update the process. mapper_id (int): ID of the mapper record to update. is_designer (bool): Whether the caller has designer permissions; used for authorization updates. Returns: dict: A dictionary containing any of the updated sections, with keys among { "formData", "mapper", "authorizations", "process" } depending on input. Raises: BusinessException: If required identifiers are missing or entities are invalid. """ # Validate that at least one updatable section(with keys) is present expected_keys = ("formData", "mapper", "authorizations", "process") self.validate_request_data(request_data, expected_keys) response = {} # Registry of handler functions for different data types handlers = { "formData": self.handle_form_data, "mapper": self.handle_mapper_data, "authorizations": self.handle_authorization_data, "process": self.handle_process_data, } # Common context for all handlers handler_context = {"mapper_id": mapper_id, "is_designer": is_designer} # Process each supported data type present in the request # Note: Update order matters # Updating process last avoids autoflush errors caused by transient str XML in # the Process entity before it's converted to bytes. for key in ("formData", "mapper", "authorizations", "process"): data = request_data.get(key) if data is not None: try: response[key] = handlers[key](data, **handler_context) except Exception as e: current_app.logger.error(f"Error processing {key}: {str(e)}") raise return response