From 6f1510c4f35148e6f975c0c7ab41fa7bb4caa5b3 Mon Sep 17 00:00:00 2001 From: jasquat Date: Thu, 29 Dec 2022 16:04:59 -0500 Subject: [PATCH] moved some more apis w/ burnettk --- .../routes/process_api_blueprint.py | 188 ------------------ .../routes/process_groups_controller.py | 7 +- .../routes/process_instances_controller.py | 11 +- .../routes/process_models_controller.py | 19 +- .../routes/script_unit_tests_controller.py | 145 ++++++++++++++ .../routes/secrets_controller.py | 1 + .../routes/tasks_controller.py | 30 ++- 7 files changed, 202 insertions(+), 199 deletions(-) create mode 100644 spiffworkflow-backend/src/spiffworkflow_backend/routes/script_unit_tests_controller.py diff --git a/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_api_blueprint.py b/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_api_blueprint.py index 5aa8d9ad1..3800fbb86 100644 --- a/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_api_blueprint.py +++ b/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_api_blueprint.py @@ -1,8 +1,5 @@ """APIs for dealing with process groups, process models, and process instances.""" import json -import os -import random -import string import uuid from typing import Any from typing import Dict @@ -12,7 +9,6 @@ from typing import Union import connexion # type: ignore import flask.wrappers -import jinja2 import werkzeug from flask import Blueprint from flask import current_app @@ -27,79 +23,35 @@ from flask_bpmn.models.db import db from lxml import etree # type: ignore from lxml.builder import ElementMaker # type: ignore from SpiffWorkflow.task import Task as SpiffTask # type: ignore -from SpiffWorkflow.task import TaskState from sqlalchemy import and_ -from sqlalchemy import asc -from sqlalchemy import desc from sqlalchemy import or_ from spiffworkflow_backend.exceptions.process_entity_not_found_error import ( ProcessEntityNotFoundError, ) -from spiffworkflow_backend.models.file import FileSchema -from spiffworkflow_backend.models.group import GroupModel from spiffworkflow_backend.models.human_task import HumanTaskModel from spiffworkflow_backend.models.human_task_user import HumanTaskUserModel -from spiffworkflow_backend.models.message_correlation import MessageCorrelationModel -from spiffworkflow_backend.models.message_instance import MessageInstanceModel -from spiffworkflow_backend.models.message_model import MessageModel -from spiffworkflow_backend.models.message_triggerable_process_model import ( - MessageTriggerableProcessModel, -) from spiffworkflow_backend.models.principal import PrincipalModel -from spiffworkflow_backend.models.process_group import ProcessGroup -from spiffworkflow_backend.models.process_group import ProcessGroupSchema -from spiffworkflow_backend.models.process_instance import ProcessInstanceApiSchema -from spiffworkflow_backend.models.process_instance import ( - ProcessInstanceCannotBeDeletedError, -) from spiffworkflow_backend.models.process_instance import ProcessInstanceModel from spiffworkflow_backend.models.process_instance import ProcessInstanceModelSchema -from spiffworkflow_backend.models.process_instance import ProcessInstanceStatus from spiffworkflow_backend.models.process_instance import ( ProcessInstanceTaskDataCannotBeUpdatedError, ) -from spiffworkflow_backend.models.process_instance_metadata import ( - ProcessInstanceMetadataModel, -) -from spiffworkflow_backend.models.process_instance_report import ( - ProcessInstanceReportModel, -) from spiffworkflow_backend.models.process_model import ProcessModelInfo -from spiffworkflow_backend.models.process_model import ProcessModelInfoSchema -from spiffworkflow_backend.models.secret_model import SecretModel -from spiffworkflow_backend.models.secret_model import SecretModelSchema from spiffworkflow_backend.models.spec_reference import SpecReferenceCache from spiffworkflow_backend.models.spec_reference import SpecReferenceNotFoundError from spiffworkflow_backend.models.spec_reference import SpecReferenceSchema -from spiffworkflow_backend.models.spiff_logging import SpiffLoggingModel -from spiffworkflow_backend.models.spiff_step_details import SpiffStepDetailsModel -from spiffworkflow_backend.models.user import UserModel from spiffworkflow_backend.routes.user import verify_token from spiffworkflow_backend.services.authorization_service import AuthorizationService -from spiffworkflow_backend.services.error_handling_service import ErrorHandlingService -from spiffworkflow_backend.services.file_system_service import FileSystemService from spiffworkflow_backend.services.git_service import GitCommandError from spiffworkflow_backend.services.git_service import GitService -from spiffworkflow_backend.services.message_service import MessageService from spiffworkflow_backend.services.process_instance_processor import ( ProcessInstanceProcessor, ) -from spiffworkflow_backend.services.process_instance_report_service import ( - ProcessInstanceReportFilter, -) -from spiffworkflow_backend.services.process_instance_report_service import ( - ProcessInstanceReportService, -) -from spiffworkflow_backend.services.process_instance_service import ( - ProcessInstanceService, -) from spiffworkflow_backend.services.process_model_service import ProcessModelService -from spiffworkflow_backend.services.script_unit_test_runner import ScriptUnitTestRunner from spiffworkflow_backend.services.secret_service import SecretService from spiffworkflow_backend.services.service_task_service import ServiceTaskService from spiffworkflow_backend.services.spec_file_service import SpecFileService -from spiffworkflow_backend.services.user_service import UserService class TaskDataSelectOption(TypedDict): @@ -235,115 +187,6 @@ def process_data_show( ) -def script_unit_test_create( - modified_process_model_identifier: str, body: Dict[str, Union[str, bool, int]] -) -> flask.wrappers.Response: - """Script_unit_test_create.""" - bpmn_task_identifier = _get_required_parameter_or_raise( - "bpmn_task_identifier", body - ) - input_json = _get_required_parameter_or_raise("input_json", body) - expected_output_json = _get_required_parameter_or_raise( - "expected_output_json", body - ) - - process_model_identifier = modified_process_model_identifier.replace(":", "/") - process_model = _get_process_model(process_model_identifier) - file = SpecFileService.get_files(process_model, process_model.primary_file_name)[0] - if file is None: - raise ApiError( - error_code="cannot_find_file", - message=f"Could not find the primary bpmn file for process_model: {process_model.id}", - status_code=404, - ) - - # TODO: move this to an xml service or something - file_contents = SpecFileService.get_data(process_model, file.name) - bpmn_etree_element = etree.fromstring(file_contents) - - nsmap = bpmn_etree_element.nsmap - spiff_element_maker = ElementMaker( - namespace="http://spiffworkflow.org/bpmn/schema/1.0/core", nsmap=nsmap - ) - - script_task_elements = bpmn_etree_element.xpath( - f"//bpmn:scriptTask[@id='{bpmn_task_identifier}']", - namespaces={"bpmn": "http://www.omg.org/spec/BPMN/20100524/MODEL"}, - ) - if len(script_task_elements) == 0: - raise ApiError( - error_code="missing_script_task", - message=f"Cannot find a script task with id: {bpmn_task_identifier}", - status_code=404, - ) - script_task_element = script_task_elements[0] - - extension_elements = None - extension_elements_array = script_task_element.xpath( - ".//bpmn:extensionElements", - namespaces={"bpmn": "http://www.omg.org/spec/BPMN/20100524/MODEL"}, - ) - if len(extension_elements_array) == 0: - bpmn_element_maker = ElementMaker( - namespace="http://www.omg.org/spec/BPMN/20100524/MODEL", nsmap=nsmap - ) - extension_elements = bpmn_element_maker("extensionElements") - script_task_element.append(extension_elements) - else: - extension_elements = extension_elements_array[0] - - unit_test_elements = None - unit_test_elements_array = extension_elements.xpath( - "//spiffworkflow:unitTests", - namespaces={"spiffworkflow": "http://spiffworkflow.org/bpmn/schema/1.0/core"}, - ) - if len(unit_test_elements_array) == 0: - unit_test_elements = spiff_element_maker("unitTests") - extension_elements.append(unit_test_elements) - else: - unit_test_elements = unit_test_elements_array[0] - - fuzz = "".join( - random.choice(string.ascii_uppercase + string.digits) # noqa: S311 - for _ in range(7) - ) - unit_test_id = f"unit_test_{fuzz}" - - input_json_element = spiff_element_maker("inputJson", json.dumps(input_json)) - expected_output_json_element = spiff_element_maker( - "expectedOutputJson", json.dumps(expected_output_json) - ) - unit_test_element = spiff_element_maker("unitTest", id=unit_test_id) - unit_test_element.append(input_json_element) - unit_test_element.append(expected_output_json_element) - unit_test_elements.append(unit_test_element) - SpecFileService.update_file( - process_model, file.name, etree.tostring(bpmn_etree_element) - ) - - return Response(json.dumps({"ok": True}), status=202, mimetype="application/json") - - -def script_unit_test_run( - modified_process_model_identifier: str, body: Dict[str, Union[str, bool, int]] -) -> flask.wrappers.Response: - """Script_unit_test_run.""" - # FIXME: We should probably clear this somewhere else but this works - current_app.config["THREAD_LOCAL_DATA"].process_instance_id = None - current_app.config["THREAD_LOCAL_DATA"].spiff_step = None - - python_script = _get_required_parameter_or_raise("python_script", body) - input_json = _get_required_parameter_or_raise("input_json", body) - expected_output_json = _get_required_parameter_or_raise( - "expected_output_json", body - ) - - result = ScriptUnitTestRunner.run_with_script_and_pre_post_contexts( - python_script, input_json, expected_output_json - ) - return make_response(jsonify(result), 200) - - def _find_principal_or_raise() -> PrincipalModel: """Find_principal_or_raise.""" principal = PrincipalModel.query.filter_by(user_id=g.user.id).first() @@ -358,37 +201,6 @@ def _find_principal_or_raise() -> PrincipalModel: return principal # type: ignore -def get_value_from_array_with_index(array: list, index: int) -> Any: - """Get_value_from_array_with_index.""" - if index < 0: - return None - - if index >= len(array): - return None - - return array[index] - - -def prepare_form_data( - form_file: str, task_data: Union[dict, None], process_model: ProcessModelInfo -) -> str: - """Prepare_form_data.""" - if task_data is None: - return "" - - file_contents = SpecFileService.get_data(process_model, form_file).decode("utf-8") - return render_jinja_template(file_contents, task_data) - - -def render_jinja_template(unprocessed_template: str, data: dict[str, Any]) -> str: - """Render_jinja_template.""" - jinja_environment = jinja2.Environment( - autoescape=True, lstrip_blocks=True, trim_blocks=True - ) - template = jinja_environment.from_string(unprocessed_template) - return template.render(**data) - - # sample body: # {"ref": "refs/heads/main", "repository": {"name": "sample-process-models", # "full_name": "sartography/sample-process-models", "private": False .... }} diff --git a/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_groups_controller.py b/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_groups_controller.py index e96c1b482..b0d43a4c9 100644 --- a/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_groups_controller.py +++ b/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_groups_controller.py @@ -2,10 +2,8 @@ import json from typing import Any from typing import Optional -from typing import TypedDict import flask.wrappers -from flask import Blueprint from flask import g from flask import jsonify from flask import make_response @@ -17,8 +15,11 @@ from spiffworkflow_backend.exceptions.process_entity_not_found_error import ( ) from spiffworkflow_backend.models.process_group import ProcessGroup from spiffworkflow_backend.models.process_group import ProcessGroupSchema +from spiffworkflow_backend.routes.process_api_blueprint import _commit_and_push_to_git +from spiffworkflow_backend.routes.process_api_blueprint import ( + _un_modify_modified_process_model_id, +) from spiffworkflow_backend.services.process_model_service import ProcessModelService -from spiffworkflow_backend.routes.process_api_blueprint import _commit_and_push_to_git, _un_modify_modified_process_model_id def process_group_create(body: dict) -> flask.wrappers.Response: diff --git a/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_instances_controller.py b/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_instances_controller.py index f28c36c78..f83ae11fb 100644 --- a/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_instances_controller.py +++ b/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_instances_controller.py @@ -1,5 +1,4 @@ """APIs for dealing with process groups, process models, and process instances.""" -from spiffworkflow_backend.routes.process_api_blueprint import _find_process_instance_for_me_or_raise, _find_process_instance_by_id_or_raise, _un_modify_modified_process_model_id, _get_process_instance import json from typing import Any from typing import Dict @@ -31,6 +30,16 @@ from spiffworkflow_backend.models.process_instance_report import ( from spiffworkflow_backend.models.spiff_logging import SpiffLoggingModel from spiffworkflow_backend.models.spiff_step_details import SpiffStepDetailsModel from spiffworkflow_backend.models.user import UserModel +from spiffworkflow_backend.routes.process_api_blueprint import ( + _find_process_instance_by_id_or_raise, +) +from spiffworkflow_backend.routes.process_api_blueprint import ( + _find_process_instance_for_me_or_raise, +) +from spiffworkflow_backend.routes.process_api_blueprint import _get_process_instance +from spiffworkflow_backend.routes.process_api_blueprint import ( + _un_modify_modified_process_model_id, +) from spiffworkflow_backend.services.error_handling_service import ErrorHandlingService from spiffworkflow_backend.services.message_service import MessageService from spiffworkflow_backend.services.process_instance_processor import ( diff --git a/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_models_controller.py b/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_models_controller.py index e0699cbe7..182c8c8bb 100644 --- a/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_models_controller.py +++ b/spiffworkflow-backend/src/spiffworkflow_backend/routes/process_models_controller.py @@ -1,6 +1,5 @@ """APIs for dealing with process groups, process models, and process instances.""" import json -from spiffworkflow_backend.routes.process_api_blueprint import _commit_and_push_to_git, _un_modify_modified_process_model_id, _get_process_model, _get_file_from_request from typing import Any from typing import Dict from typing import Optional @@ -17,8 +16,14 @@ from flask_bpmn.api.api_error import ApiError from spiffworkflow_backend.models.file import FileSchema from spiffworkflow_backend.models.process_model import ProcessModelInfo from spiffworkflow_backend.models.process_model import ProcessModelInfoSchema -from spiffworkflow_backend.services.git_service import MissingGitConfigsError +from spiffworkflow_backend.routes.process_api_blueprint import _commit_and_push_to_git +from spiffworkflow_backend.routes.process_api_blueprint import _get_file_from_request +from spiffworkflow_backend.routes.process_api_blueprint import _get_process_model +from spiffworkflow_backend.routes.process_api_blueprint import ( + _un_modify_modified_process_model_id, +) from spiffworkflow_backend.services.git_service import GitService +from spiffworkflow_backend.services.git_service import MissingGitConfigsError from spiffworkflow_backend.services.process_model_service import ProcessModelService from spiffworkflow_backend.services.spec_file_service import SpecFileService @@ -247,7 +252,10 @@ def process_model_file_delete( return Response(json.dumps({"ok": True}), status=200, mimetype="application/json") -def process_model_file_create(modified_process_model_identifier: str) -> flask.wrappers.Response: +def process_model_file_create( + modified_process_model_identifier: str, +) -> flask.wrappers.Response: + """Process_model_file_create.""" process_model_identifier = modified_process_model_identifier.replace(":", "/") process_model = _get_process_model(process_model_identifier) request_file = _get_file_from_request() @@ -272,7 +280,10 @@ def process_model_file_create(modified_process_model_identifier: str) -> flask.w ) -def process_model_file_show(modified_process_model_identifier: str, file_name: str) -> Any: +def process_model_file_show( + modified_process_model_identifier: str, file_name: str +) -> Any: + """Process_model_file_show.""" process_model_identifier = modified_process_model_identifier.replace(":", "/") process_model = _get_process_model(process_model_identifier) files = SpecFileService.get_files(process_model, file_name) diff --git a/spiffworkflow-backend/src/spiffworkflow_backend/routes/script_unit_tests_controller.py b/spiffworkflow-backend/src/spiffworkflow_backend/routes/script_unit_tests_controller.py new file mode 100644 index 000000000..835ef9a95 --- /dev/null +++ b/spiffworkflow-backend/src/spiffworkflow_backend/routes/script_unit_tests_controller.py @@ -0,0 +1,145 @@ +"""APIs for dealing with process groups, process models, and process instances.""" +import json +from spiffworkflow_backend.routes.process_api_blueprint import _get_required_parameter_or_raise +from spiffworkflow_backend.routes.process_api_blueprint import _get_process_model +import random +import string +from typing import Dict +from typing import Union + +import flask.wrappers +from flask import current_app +from flask import g +from flask import jsonify +from flask import make_response +from flask.wrappers import Response +from flask_bpmn.api.api_error import ApiError +from lxml import etree # type: ignore +from lxml.builder import ElementMaker # type: ignore + +from spiffworkflow_backend.models.principal import PrincipalModel +from spiffworkflow_backend.services.script_unit_test_runner import ScriptUnitTestRunner +from spiffworkflow_backend.services.spec_file_service import SpecFileService + + +def script_unit_test_create( + modified_process_model_identifier: str, body: Dict[str, Union[str, bool, int]] +) -> flask.wrappers.Response: + """Script_unit_test_create.""" + bpmn_task_identifier = _get_required_parameter_or_raise( + "bpmn_task_identifier", body + ) + input_json = _get_required_parameter_or_raise("input_json", body) + expected_output_json = _get_required_parameter_or_raise( + "expected_output_json", body + ) + + process_model_identifier = modified_process_model_identifier.replace(":", "/") + process_model = _get_process_model(process_model_identifier) + file = SpecFileService.get_files(process_model, process_model.primary_file_name)[0] + if file is None: + raise ApiError( + error_code="cannot_find_file", + message=f"Could not find the primary bpmn file for process_model: {process_model.id}", + status_code=404, + ) + + # TODO: move this to an xml service or something + file_contents = SpecFileService.get_data(process_model, file.name) + bpmn_etree_element = etree.fromstring(file_contents) + + nsmap = bpmn_etree_element.nsmap + spiff_element_maker = ElementMaker( + namespace="http://spiffworkflow.org/bpmn/schema/1.0/core", nsmap=nsmap + ) + + script_task_elements = bpmn_etree_element.xpath( + f"//bpmn:scriptTask[@id='{bpmn_task_identifier}']", + namespaces={"bpmn": "http://www.omg.org/spec/BPMN/20100524/MODEL"}, + ) + if len(script_task_elements) == 0: + raise ApiError( + error_code="missing_script_task", + message=f"Cannot find a script task with id: {bpmn_task_identifier}", + status_code=404, + ) + script_task_element = script_task_elements[0] + + extension_elements = None + extension_elements_array = script_task_element.xpath( + ".//bpmn:extensionElements", + namespaces={"bpmn": "http://www.omg.org/spec/BPMN/20100524/MODEL"}, + ) + if len(extension_elements_array) == 0: + bpmn_element_maker = ElementMaker( + namespace="http://www.omg.org/spec/BPMN/20100524/MODEL", nsmap=nsmap + ) + extension_elements = bpmn_element_maker("extensionElements") + script_task_element.append(extension_elements) + else: + extension_elements = extension_elements_array[0] + + unit_test_elements = None + unit_test_elements_array = extension_elements.xpath( + "//spiffworkflow:unitTests", + namespaces={"spiffworkflow": "http://spiffworkflow.org/bpmn/schema/1.0/core"}, + ) + if len(unit_test_elements_array) == 0: + unit_test_elements = spiff_element_maker("unitTests") + extension_elements.append(unit_test_elements) + else: + unit_test_elements = unit_test_elements_array[0] + + fuzz = "".join( + random.choice(string.ascii_uppercase + string.digits) # noqa: S311 + for _ in range(7) + ) + unit_test_id = f"unit_test_{fuzz}" + + input_json_element = spiff_element_maker("inputJson", json.dumps(input_json)) + expected_output_json_element = spiff_element_maker( + "expectedOutputJson", json.dumps(expected_output_json) + ) + unit_test_element = spiff_element_maker("unitTest", id=unit_test_id) + unit_test_element.append(input_json_element) + unit_test_element.append(expected_output_json_element) + unit_test_elements.append(unit_test_element) + SpecFileService.update_file( + process_model, file.name, etree.tostring(bpmn_etree_element) + ) + + return Response(json.dumps({"ok": True}), status=202, mimetype="application/json") + + +def script_unit_test_run( + modified_process_model_identifier: str, body: Dict[str, Union[str, bool, int]] +) -> flask.wrappers.Response: + """Script_unit_test_run.""" + # FIXME: We should probably clear this somewhere else but this works + current_app.config["THREAD_LOCAL_DATA"].process_instance_id = None + current_app.config["THREAD_LOCAL_DATA"].spiff_step = None + + python_script = _get_required_parameter_or_raise("python_script", body) + input_json = _get_required_parameter_or_raise("input_json", body) + expected_output_json = _get_required_parameter_or_raise( + "expected_output_json", body + ) + + result = ScriptUnitTestRunner.run_with_script_and_pre_post_contexts( + python_script, input_json, expected_output_json + ) + return make_response(jsonify(result), 200) + + +def _find_principal_or_raise() -> PrincipalModel: + """Find_principal_or_raise.""" + principal = PrincipalModel.query.filter_by(user_id=g.user.id).first() + if principal is None: + raise ( + ApiError( + error_code="principal_not_found", + message=f"Principal not found from user id: {g.user.id}", + status_code=400, + ) + ) + return principal # type: ignore diff --git a/spiffworkflow-backend/src/spiffworkflow_backend/routes/secrets_controller.py b/spiffworkflow-backend/src/spiffworkflow_backend/routes/secrets_controller.py index 13b58e8f3..fdf4c7fae 100644 --- a/spiffworkflow-backend/src/spiffworkflow_backend/routes/secrets_controller.py +++ b/spiffworkflow-backend/src/spiffworkflow_backend/routes/secrets_controller.py @@ -16,6 +16,7 @@ from spiffworkflow_backend.services.user_service import UserService def secret_show(key: str) -> Optional[str]: + """Secret_show.""" return SecretService.get_secret(key) diff --git a/spiffworkflow-backend/src/spiffworkflow_backend/routes/tasks_controller.py b/spiffworkflow-backend/src/spiffworkflow_backend/routes/tasks_controller.py index 455e5d3b4..1829dcd35 100644 --- a/spiffworkflow-backend/src/spiffworkflow_backend/routes/tasks_controller.py +++ b/spiffworkflow-backend/src/spiffworkflow_backend/routes/tasks_controller.py @@ -1,11 +1,14 @@ """APIs for dealing with process groups, process models, and process instances.""" import json +from spiffworkflow_backend.routes.process_api_blueprint import _get_process_model, _get_required_parameter_or_raise import os from typing import Any from typing import Dict from typing import Optional +from typing import Union import flask.wrappers +import jinja2 from flask import g from flask import jsonify from flask import make_response @@ -21,6 +24,7 @@ from spiffworkflow_backend.models.human_task import HumanTaskModel from spiffworkflow_backend.models.human_task_user import HumanTaskUserModel from spiffworkflow_backend.models.process_instance import ProcessInstanceModel from spiffworkflow_backend.models.process_instance import ProcessInstanceStatus +from spiffworkflow_backend.models.process_model import ProcessModelInfo from spiffworkflow_backend.models.user import UserModel from spiffworkflow_backend.services.authorization_service import AuthorizationService from spiffworkflow_backend.services.file_system_service import FileSystemService @@ -164,7 +168,7 @@ def task_show(process_instance_id: int, task_id: str) -> flask.wrappers.Response ) ) - form_contents = prepare_form_data( + form_contents = _prepare_form_data( form_schema_file_name, task.data, process_model_with_form, @@ -189,7 +193,7 @@ def task_show(process_instance_id: int, task_id: str) -> flask.wrappers.Response task.form_schema = form_dict if form_ui_schema_file_name: - ui_form_contents = prepare_form_data( + ui_form_contents = _prepare_form_data( form_ui_schema_file_name, task.data, process_model_with_form, @@ -199,7 +203,7 @@ def task_show(process_instance_id: int, task_id: str) -> flask.wrappers.Response if task.properties and task.data and "instructionsForEndUser" in task.properties: if task.properties["instructionsForEndUser"]: - task.properties["instructionsForEndUser"] = render_jinja_template( + task.properties["instructionsForEndUser"] = _render_jinja_template( task.properties["instructionsForEndUser"], task.data ) return make_response(jsonify(task), 200) @@ -394,3 +398,23 @@ def _get_tasks( }, } return make_response(jsonify(response_json), 200) + + +def _prepare_form_data( + form_file: str, task_data: Union[dict, None], process_model: ProcessModelInfo +) -> str: + """Prepare_form_data.""" + if task_data is None: + return "" + + file_contents = SpecFileService.get_data(process_model, form_file).decode("utf-8") + return _render_jinja_template(file_contents, task_data) + + +def _render_jinja_template(unprocessed_template: str, data: dict[str, Any]) -> str: + """Render_jinja_template.""" + jinja_environment = jinja2.Environment( + autoescape=True, lstrip_blocks=True, trim_blocks=True + ) + template = jinja_environment.from_string(unprocessed_template) + return template.render(**data)