moved some more apis w/ burnettk

This commit is contained in:
jasquat 2022-12-29 16:04:59 -05:00
parent 1c3da92f81
commit 6f1510c4f3
7 changed files with 202 additions and 199 deletions

View File

@ -1,8 +1,5 @@
"""APIs for dealing with process groups, process models, and process instances.""" """APIs for dealing with process groups, process models, and process instances."""
import json import json
import os
import random
import string
import uuid import uuid
from typing import Any from typing import Any
from typing import Dict from typing import Dict
@ -12,7 +9,6 @@ from typing import Union
import connexion # type: ignore import connexion # type: ignore
import flask.wrappers import flask.wrappers
import jinja2
import werkzeug import werkzeug
from flask import Blueprint from flask import Blueprint
from flask import current_app from flask import current_app
@ -27,79 +23,35 @@ from flask_bpmn.models.db import db
from lxml import etree # type: ignore from lxml import etree # type: ignore
from lxml.builder import ElementMaker # type: ignore from lxml.builder import ElementMaker # type: ignore
from SpiffWorkflow.task import Task as SpiffTask # type: ignore from SpiffWorkflow.task import Task as SpiffTask # type: ignore
from SpiffWorkflow.task import TaskState
from sqlalchemy import and_ from sqlalchemy import and_
from sqlalchemy import asc
from sqlalchemy import desc
from sqlalchemy import or_ from sqlalchemy import or_
from spiffworkflow_backend.exceptions.process_entity_not_found_error import ( from spiffworkflow_backend.exceptions.process_entity_not_found_error import (
ProcessEntityNotFoundError, 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 import HumanTaskModel
from spiffworkflow_backend.models.human_task_user import HumanTaskUserModel 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.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 ProcessInstanceModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceModelSchema from spiffworkflow_backend.models.process_instance import ProcessInstanceModelSchema
from spiffworkflow_backend.models.process_instance import ProcessInstanceStatus
from spiffworkflow_backend.models.process_instance import ( from spiffworkflow_backend.models.process_instance import (
ProcessInstanceTaskDataCannotBeUpdatedError, 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 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 SpecReferenceCache
from spiffworkflow_backend.models.spec_reference import SpecReferenceNotFoundError from spiffworkflow_backend.models.spec_reference import SpecReferenceNotFoundError
from spiffworkflow_backend.models.spec_reference import SpecReferenceSchema 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.routes.user import verify_token
from spiffworkflow_backend.services.authorization_service import AuthorizationService 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 GitCommandError
from spiffworkflow_backend.services.git_service import GitService from spiffworkflow_backend.services.git_service import GitService
from spiffworkflow_backend.services.message_service import MessageService
from spiffworkflow_backend.services.process_instance_processor import ( from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor, 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.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.secret_service import SecretService
from spiffworkflow_backend.services.service_task_service import ServiceTaskService from spiffworkflow_backend.services.service_task_service import ServiceTaskService
from spiffworkflow_backend.services.spec_file_service import SpecFileService from spiffworkflow_backend.services.spec_file_service import SpecFileService
from spiffworkflow_backend.services.user_service import UserService
class TaskDataSelectOption(TypedDict): 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: def _find_principal_or_raise() -> PrincipalModel:
"""Find_principal_or_raise.""" """Find_principal_or_raise."""
principal = PrincipalModel.query.filter_by(user_id=g.user.id).first() 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 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: # sample body:
# {"ref": "refs/heads/main", "repository": {"name": "sample-process-models", # {"ref": "refs/heads/main", "repository": {"name": "sample-process-models",
# "full_name": "sartography/sample-process-models", "private": False .... }} # "full_name": "sartography/sample-process-models", "private": False .... }}

View File

@ -2,10 +2,8 @@
import json import json
from typing import Any from typing import Any
from typing import Optional from typing import Optional
from typing import TypedDict
import flask.wrappers import flask.wrappers
from flask import Blueprint
from flask import g from flask import g
from flask import jsonify from flask import jsonify
from flask import make_response 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 ProcessGroup
from spiffworkflow_backend.models.process_group import ProcessGroupSchema 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.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: def process_group_create(body: dict) -> flask.wrappers.Response:

View File

@ -1,5 +1,4 @@
"""APIs for dealing with process groups, process models, and process instances.""" """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 import json
from typing import Any from typing import Any
from typing import Dict 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_logging import SpiffLoggingModel
from spiffworkflow_backend.models.spiff_step_details import SpiffStepDetailsModel from spiffworkflow_backend.models.spiff_step_details import SpiffStepDetailsModel
from spiffworkflow_backend.models.user import UserModel 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.error_handling_service import ErrorHandlingService
from spiffworkflow_backend.services.message_service import MessageService from spiffworkflow_backend.services.message_service import MessageService
from spiffworkflow_backend.services.process_instance_processor import ( from spiffworkflow_backend.services.process_instance_processor import (

View File

@ -1,6 +1,5 @@
"""APIs for dealing with process groups, process models, and process instances.""" """APIs for dealing with process groups, process models, and process instances."""
import json 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 Any
from typing import Dict from typing import Dict
from typing import Optional 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.file import FileSchema
from spiffworkflow_backend.models.process_model import ProcessModelInfo from spiffworkflow_backend.models.process_model import ProcessModelInfo
from spiffworkflow_backend.models.process_model import ProcessModelInfoSchema 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 GitService
from spiffworkflow_backend.services.git_service import MissingGitConfigsError
from spiffworkflow_backend.services.process_model_service import ProcessModelService from spiffworkflow_backend.services.process_model_service import ProcessModelService
from spiffworkflow_backend.services.spec_file_service import SpecFileService 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") 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_identifier = modified_process_model_identifier.replace(":", "/")
process_model = _get_process_model(process_model_identifier) process_model = _get_process_model(process_model_identifier)
request_file = _get_file_from_request() 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_identifier = modified_process_model_identifier.replace(":", "/")
process_model = _get_process_model(process_model_identifier) process_model = _get_process_model(process_model_identifier)
files = SpecFileService.get_files(process_model, file_name) files = SpecFileService.get_files(process_model, file_name)

View File

@ -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

View File

@ -16,6 +16,7 @@ from spiffworkflow_backend.services.user_service import UserService
def secret_show(key: str) -> Optional[str]: def secret_show(key: str) -> Optional[str]:
"""Secret_show."""
return SecretService.get_secret(key) return SecretService.get_secret(key)

View File

@ -1,11 +1,14 @@
"""APIs for dealing with process groups, process models, and process instances.""" """APIs for dealing with process groups, process models, and process instances."""
import json import json
from spiffworkflow_backend.routes.process_api_blueprint import _get_process_model, _get_required_parameter_or_raise
import os import os
from typing import Any from typing import Any
from typing import Dict from typing import Dict
from typing import Optional from typing import Optional
from typing import Union
import flask.wrappers import flask.wrappers
import jinja2
from flask import g from flask import g
from flask import jsonify from flask import jsonify
from flask import make_response 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.human_task_user import HumanTaskUserModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from spiffworkflow_backend.models.process_instance import ProcessInstanceStatus 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.models.user import UserModel
from spiffworkflow_backend.services.authorization_service import AuthorizationService from spiffworkflow_backend.services.authorization_service import AuthorizationService
from spiffworkflow_backend.services.file_system_service import FileSystemService 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, form_schema_file_name,
task.data, task.data,
process_model_with_form, 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 task.form_schema = form_dict
if form_ui_schema_file_name: if form_ui_schema_file_name:
ui_form_contents = prepare_form_data( ui_form_contents = _prepare_form_data(
form_ui_schema_file_name, form_ui_schema_file_name,
task.data, task.data,
process_model_with_form, 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 and task.data and "instructionsForEndUser" in task.properties:
if task.properties["instructionsForEndUser"]: if task.properties["instructionsForEndUser"]:
task.properties["instructionsForEndUser"] = render_jinja_template( task.properties["instructionsForEndUser"] = _render_jinja_template(
task.properties["instructionsForEndUser"], task.data task.properties["instructionsForEndUser"], task.data
) )
return make_response(jsonify(task), 200) return make_response(jsonify(task), 200)
@ -394,3 +398,23 @@ def _get_tasks(
}, },
} }
return make_response(jsonify(response_json), 200) 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)