diff --git a/src/spiffworkflow_backend/api.yml b/src/spiffworkflow_backend/api.yml index 55653fa1..304fa4e1 100755 --- a/src/spiffworkflow_backend/api.yml +++ b/src/spiffworkflow_backend/api.yml @@ -1060,6 +1060,39 @@ paths: schema: $ref: "#/components/schemas/OkTrue" + /process-instance-reset/{modified_process_model_identifier}/{process_instance_id}/{spiff_step}: + parameters: + - name: modified_process_model_identifier + in: path + required: true + description: The modified process model id + schema: + type: string + - name: process_instance_id + in: path + required: true + description: The unique id of an existing process instance. + schema: + type: integer + - name: spiff_step + in: query + required: false + description: Reset the process to this state + schema: + type: integer + post: + operationId: spiffworkflow_backend.routes.process_api_blueprint.process_instance_reset + summary: Reset a process instance to an earlier step + tags: + - Process Instances + responses: + "200": + description: Empty ok true response on successful resume. + content: + application/json: + schema: + $ref: "#/components/schemas/OkTrue" + /process-instances/reports: parameters: - name: page @@ -1484,6 +1517,66 @@ paths: schema: $ref: "#/components/schemas/Workflow" + /send-event/{modified_process_model_identifier}/{process_instance_id}: + parameters: + - name: modified_process_model_identifier + in: path + required: true + description: The modified id of an existing process model + schema: + type: string + - name: process_instance_id + in: path + required: true + description: The unique id of the process instance + schema: + type: string + post: + operationId: spiffworkflow_backend.routes.process_api_blueprint.send_bpmn_event + summary: Send a BPMN event to the process + tags: + - Process Instances + responses: + "200": + description: Event Sent Successfully + content: + application/json: + schema: + $ref: "#/components/schemas/Workflow" + + /task-complete/{modified_process_model_identifier}/{process_instance_id}/{task_id}: + parameters: + - name: modified_process_model_identifier + in: path + required: true + description: The modified id of an existing process model + schema: + type: string + - name: process_instance_id + in: path + required: true + description: The unique id of the process instance + schema: + type: string + - name: task_id + in: path + required: true + description: The unique id of the task. + schema: + type: string + post: + operationId: spiffworkflow_backend.routes.process_api_blueprint.manual_complete_task + summary: Mark a task complete without executing it + tags: + - Process Instances + responses: + "200": + description: Event Sent Successfully + content: + application/json: + schema: + $ref: "#/components/schemas/Workflow" + /service-tasks: get: tags: diff --git a/src/spiffworkflow_backend/models/task.py b/src/spiffworkflow_backend/models/task.py index 60deda84..896c79e1 100644 --- a/src/spiffworkflow_backend/models/task.py +++ b/src/spiffworkflow_backend/models/task.py @@ -118,6 +118,7 @@ class Task: form_schema: Union[str, None] = None, form_ui_schema: Union[str, None] = None, parent: Optional[str] = None, + event_definition: Union[dict[str, Any], None] = None, call_activity_process_identifier: Optional[str] = None, ): """__init__.""" @@ -130,6 +131,7 @@ class Task: self.documentation = documentation self.lane = lane self.parent = parent + self.event_definition = event_definition self.call_activity_process_identifier = call_activity_process_identifier self.data = data @@ -189,6 +191,7 @@ class Task: "form_schema": self.form_schema, "form_ui_schema": self.form_ui_schema, "parent": self.parent, + "event_definition": self.event_definition, "call_activity_process_identifier": self.call_activity_process_identifier, } @@ -290,6 +293,7 @@ class TaskSchema(Schema): "process_instance_id", "form_schema", "form_ui_schema", + "event_definition", ] multi_instance_type = EnumField(MultiInstanceType) diff --git a/src/spiffworkflow_backend/routes/process_api_blueprint.py b/src/spiffworkflow_backend/routes/process_api_blueprint.py index 68133903..68259881 100644 --- a/src/spiffworkflow_backend/routes/process_api_blueprint.py +++ b/src/spiffworkflow_backend/routes/process_api_blueprint.py @@ -652,6 +652,46 @@ def process_instance_resume( return Response(json.dumps({"ok": True}), status=200, mimetype="application/json") +def process_instance_reset( + process_instance_id: int, + modified_process_model_identifier: str, + spiff_step: int = 0, +) -> flask.wrappers.Response: + """Process_instance_reset.""" + process_instance = ProcessInstanceService().get_process_instance( + process_instance_id + ) + step_detail = ( + db.session.query(SpiffStepDetailsModel) + .filter( + SpiffStepDetailsModel.process_instance_id == process_instance.id, + SpiffStepDetailsModel.spiff_step == spiff_step, + ) + .first() + ) + if step_detail is not None and process_instance.bpmn_json is not None: + bpmn_json = json.loads(process_instance.bpmn_json) + bpmn_json["tasks"] = step_detail.task_json["tasks"] + bpmn_json["subprocesses"] = step_detail.task_json["subprocesses"] + process_instance.bpmn_json = json.dumps(bpmn_json) + + db.session.add(process_instance) + try: + db.session.commit() + except Exception as e: + db.session.rollback() + raise ApiError( + error_code="reset_process_instance_error", + message=f"Could not update the Instance. Original error is {e}", + ) from e + + return Response( + json.dumps(ProcessInstanceModelSchema().dump(process_instance)), + status=200, + mimetype="application/json", + ) + + def process_instance_log_list( modified_process_model_identifier: str, process_instance_id: int, @@ -1406,7 +1446,7 @@ def process_instance_task_list( step_detail = ( db.session.query(SpiffStepDetailsModel) .filter( - SpiffStepDetailsModel.process_instance.id == process_instance.id, + SpiffStepDetailsModel.process_instance_id == process_instance.id, SpiffStepDetailsModel.spiff_step == spiff_step, ) .first() @@ -1427,7 +1467,7 @@ def process_instance_task_list( tasks = [] for spiff_task in spiff_tasks: - task = ProcessInstanceService.spiff_task_to_api_task(spiff_task) + task = ProcessInstanceService.spiff_task_to_api_task(processor, spiff_task) if get_task_data: task.data = spiff_task.data tasks.append(task) @@ -1461,7 +1501,9 @@ def task_show(process_instance_id: int, task_id: str) -> flask.wrappers.Response form_schema_file_name = properties["formJsonSchemaFilename"] if "formUiSchemaFilename" in properties: form_ui_schema_file_name = properties["formUiSchemaFilename"] - task = ProcessInstanceService.spiff_task_to_api_task(spiff_task) + + processor = ProcessInstanceProcessor(process_instance) + task = ProcessInstanceService.spiff_task_to_api_task(processor, spiff_task) task.data = spiff_task.data task.process_model_display_name = process_model.display_name task.process_model_identifier = process_model.id @@ -2077,6 +2119,107 @@ def _update_form_schema_with_task_data_as_needed( _update_form_schema_with_task_data_as_needed(o, task_data) +def update_task_data( + process_instance_id: str, + modified_process_model_identifier: str, + task_id: str, + body: Dict, +) -> Response: + """Update task data.""" + process_instance = ProcessInstanceModel.query.filter( + ProcessInstanceModel.id == int(process_instance_id) + ).first() + if process_instance: + if process_instance.status != "suspended": + raise ProcessInstanceTaskDataCannotBeUpdatedError( + f"The process instance needs to be suspended to udpate the task-data. It is currently: {process_instance.status}" + ) + + process_instance_bpmn_json_dict = json.loads(process_instance.bpmn_json) + if "new_task_data" in body: + new_task_data_str: str = body["new_task_data"] + new_task_data_dict = json.loads(new_task_data_str) + if task_id in process_instance_bpmn_json_dict["tasks"]: + process_instance_bpmn_json_dict["tasks"][task_id][ + "data" + ] = new_task_data_dict + process_instance.bpmn_json = json.dumps(process_instance_bpmn_json_dict) + db.session.add(process_instance) + try: + db.session.commit() + except Exception as e: + db.session.rollback() + raise ApiError( + error_code="update_task_data_error", + message=f"Could not update the Instance. Original error is {e}", + ) from e + else: + raise ApiError( + error_code="update_task_data_error", + message=f"Could not find Task: {task_id} in Instance: {process_instance_id}.", + ) + else: + raise ApiError( + error_code="update_task_data_error", + message=f"Could not update task data for Instance: {process_instance_id}, and Task: {task_id}.", + ) + return Response( + json.dumps(ProcessInstanceModelSchema().dump(process_instance)), + status=200, + mimetype="application/json", + ) + + +def send_bpmn_event( + modified_process_model_identifier: str, + process_instance_id: str, + body: Dict, +) -> Response: + """Send a bpmn event to a workflow.""" + process_instance = ProcessInstanceModel.query.filter( + ProcessInstanceModel.id == int(process_instance_id) + ).first() + if process_instance: + processor = ProcessInstanceProcessor(process_instance) + processor.send_bpmn_event(body) + else: + raise ApiError( + error_code="send_bpmn_event_error", + message=f"Could not send event to Instance: {process_instance_id}", + ) + return Response( + json.dumps(ProcessInstanceModelSchema().dump(process_instance)), + status=200, + mimetype="application/json", + ) + + +def manual_complete_task( + modified_process_model_identifier: str, + process_instance_id: str, + task_id: str, + body: Dict, +) -> Response: + """Mark a task complete without executing it.""" + execute = body.get("execute", True) + process_instance = ProcessInstanceModel.query.filter( + ProcessInstanceModel.id == int(process_instance_id) + ).first() + if process_instance: + processor = ProcessInstanceProcessor(process_instance) + processor.manual_complete_task(task_id, execute) + else: + raise ApiError( + error_code="complete_task", + message=f"Could not complete Task {task_id} in Instance {process_instance_id}", + ) + return Response( + json.dumps(ProcessInstanceModelSchema().dump(process_instance)), + status=200, + mimetype="application/json", + ) + + def _commit_and_push_to_git(message: str) -> None: """Commit_and_push_to_git.""" if current_app.config["GIT_COMMIT_ON_SAVE"]: diff --git a/src/spiffworkflow_backend/services/process_instance_processor.py b/src/spiffworkflow_backend/services/process_instance_processor.py index 1e548a76..f9e6c620 100644 --- a/src/spiffworkflow_backend/services/process_instance_processor.py +++ b/src/spiffworkflow_backend/services/process_instance_processor.py @@ -17,6 +17,7 @@ from typing import Optional from typing import Tuple from typing import TypedDict from typing import Union +from uuid import UUID import dateparser import pytz @@ -43,6 +44,9 @@ from SpiffWorkflow.spiff.serializer.task_spec_converters import ( CallActivityTaskConverter, ) from SpiffWorkflow.spiff.serializer.task_spec_converters import EndEventConverter +from SpiffWorkflow.spiff.serializer.task_spec_converters import ( + EventBasedGatewayConverter, +) from SpiffWorkflow.spiff.serializer.task_spec_converters import ( IntermediateCatchEventConverter, ) @@ -265,6 +269,7 @@ class ProcessInstanceProcessor: EndEventConverter, IntermediateCatchEventConverter, IntermediateThrowEventConverter, + EventBasedGatewayConverter, ManualTaskConverter, NoneTaskConverter, ReceiveTaskConverter, @@ -278,6 +283,7 @@ class ProcessInstanceProcessor: ] ) _serializer = BpmnWorkflowSerializer(wf_spec_converter, version=SERIALIZER_VERSION) + _event_serializer = EventBasedGatewayConverter() PROCESS_INSTANCE_ID_KEY = "process_instance_id" VALIDATION_PROCESS_KEY = "validate_only" @@ -616,7 +622,7 @@ class ProcessInstanceProcessor: db.session.add(pim) db.session.commit() - def save(self) -> None: + def _save(self) -> None: """Saves the current state of this processor to the database.""" self.process_instance_model.bpmn_json = self.serialize() @@ -638,6 +644,9 @@ class ProcessInstanceProcessor: db.session.add(self.process_instance_model) db.session.commit() + def save(self) -> None: + """Saves the current state and moves on to the next state.""" + self._save() human_tasks = HumanTaskModel.query.filter_by( process_instance_id=self.process_instance_model.id ).all() @@ -706,6 +715,44 @@ class ProcessInstanceProcessor: db.session.add(at) db.session.commit() + def serialize_task_spec(self, task_spec: SpiffTask) -> Any: + """Get a serialized version of a task spec.""" + # The task spec is NOT actually a SpiffTask, it is the task spec attached to a SpiffTask + # Not sure why mypy accepts this but whatever. + return self._serializer.spec_converter.convert(task_spec) + + def send_bpmn_event(self, event_data: dict[str, Any]) -> None: + """Send an event to the workflow.""" + payload = event_data.pop("payload", None) + event_definition = self._event_serializer.restore(event_data) + if payload is not None: + event_definition.payload = payload + current_app.logger.info( + f"Event of type {event_definition.event_type} sent to process instance {self.process_instance_model.id}" + ) + self.bpmn_process_instance.catch(event_definition) + self.do_engine_steps(save=True) + + def manual_complete_task(self, task_id: str, execute: bool) -> None: + """Mark the task complete optionally executing it.""" + spiff_task = self.bpmn_process_instance.get_task(UUID(task_id)) + if execute: + current_app.logger.info( + f"Manually executing Task {spiff_task.task_spec.name} of process instance {self.process_instance_model.id}" + ) + spiff_task.complete() + else: + current_app.logger.info( + f"Skipping Task {spiff_task.task_spec.name} of process instance {self.process_instance_model.id}" + ) + spiff_task._set_state(TaskState.COMPLETED) + for child in spiff_task.children: + child.task_spec._update(child) + self.bpmn_process_instance.last_task = spiff_task + self._save() + # Saving the workflow seems to reset the status + self.suspend() + @staticmethod def get_parser() -> MyCustomParser: """Get_parser.""" diff --git a/src/spiffworkflow_backend/services/process_instance_service.py b/src/spiffworkflow_backend/services/process_instance_service.py index 0dec5e44..c9722289 100644 --- a/src/spiffworkflow_backend/services/process_instance_service.py +++ b/src/spiffworkflow_backend/services/process_instance_service.py @@ -125,7 +125,7 @@ class ProcessInstanceService: if next_task_trying_again is not None: process_instance_api.next_task = ( ProcessInstanceService.spiff_task_to_api_task( - next_task_trying_again, add_docs_and_forms=True + processor, next_task_trying_again, add_docs_and_forms=True ) ) @@ -281,7 +281,9 @@ class ProcessInstanceService: @staticmethod def spiff_task_to_api_task( - spiff_task: SpiffTask, add_docs_and_forms: bool = False + processor: ProcessInstanceProcessor, + spiff_task: SpiffTask, + add_docs_and_forms: bool = False, ) -> Task: """Spiff_task_to_api_task.""" task_type = spiff_task.task_spec.spec_type @@ -315,6 +317,8 @@ class ProcessInstanceService: if spiff_task.parent: parent_id = spiff_task.parent.id + serialized_task_spec = processor.serialize_task_spec(spiff_task.task_spec) + task = Task( spiff_task.id, spiff_task.task_spec.name, @@ -328,6 +332,7 @@ class ProcessInstanceService: process_identifier=spiff_task.task_spec._wf_spec.name, properties=props, parent=parent_id, + event_definition=serialized_task_spec.get("event_definition"), call_activity_process_identifier=call_activity_process_identifier, ) diff --git a/tests/data/process_navigation/process_navigation.bpmn b/tests/data/process_navigation/process_navigation.bpmn new file mode 100644 index 00000000..9f2f26bf --- /dev/null +++ b/tests/data/process_navigation/process_navigation.bpmn @@ -0,0 +1,137 @@ + + + + + + + + Flow_1l15rbh + + + + Flow_1l15rbh + Flow_0d35i06 + Flow_0tzaigt + Flow_1vld4r2 + + + + Flow_0d35i06 + Flow_1w3n49n + + + + Flow_0tzaigt + Flow_1q47ol8 + + + + + + + Flow_1q47ol8 + + + + + Flow_1w3n49n + + + + Flow_1vld4r2 + Flow_13ai5vv + + timedelta(hours=1) + + + + + Click the button. + + Flow_13ai5vv + Flow_1vwnf3n + + + Flow_1vwnf3n + + + + + result + + + + + result + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/tests/spiffworkflow_backend/integration/test_process_api.py b/tests/spiffworkflow_backend/integration/test_process_api.py index abcc4715..77c46644 100644 --- a/tests/spiffworkflow_backend/integration/test_process_api.py +++ b/tests/spiffworkflow_backend/integration/test_process_api.py @@ -4,6 +4,7 @@ import json import os import time from typing import Any +from typing import Dict import pytest from flask.app import Flask @@ -2537,6 +2538,147 @@ class TestProcessApi(BaseTest): print("test_script_unit_test_run") + def test_send_event( + self, + app: Flask, + client: FlaskClient, + with_db_and_bpmn_file_cleanup: None, + with_super_admin_user: UserModel, + ) -> None: + """Test_script_unit_test_run.""" + process_group_id = "test_group" + process_model_id = "process_navigation" + bpmn_file_name = "process_navigation.bpmn" + bpmn_file_location = "process_navigation" + process_model_identifier = self.create_group_and_model_with_bpmn( + client=client, + user=with_super_admin_user, + process_group_id=process_group_id, + process_model_id=process_model_id, + bpmn_file_name=bpmn_file_name, + bpmn_file_location=bpmn_file_location, + ) + + bpmn_file_data_bytes = self.get_test_data_file_contents( + bpmn_file_name, bpmn_file_location + ) + self.create_spec_file( + client=client, + process_model_id=process_model_identifier, + process_model_location=process_model_identifier, + file_name=bpmn_file_name, + file_data=bpmn_file_data_bytes, + user=with_super_admin_user, + ) + + headers = self.logged_in_headers(with_super_admin_user) + response = self.create_process_instance_from_process_model_id_with_api( + client, process_model_identifier, headers + ) + process_instance_id = response.json["id"] + + client.post( + f"/v1.0/process-instances/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}/run", + headers=self.logged_in_headers(with_super_admin_user), + ) + + # This is exactly the same the test above, but some reason I to a totally irrelevant type. + data: Dict = { + "correlation_properties": [], + "expression": None, + "external": True, + "internal": False, + "payload": {"message": "message 1"}, + "name": "Message 1", + "typename": "MessageEventDefinition", + } + response = client.post( + f"/v1.0/send-event/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}", + headers=self.logged_in_headers(with_super_admin_user), + content_type="application/json", + data=json.dumps(data), + ) + assert response.json["status"] == "complete" + + response = client.get( + f"/v1.0/task-data/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}?all_tasks=true", + headers=self.logged_in_headers(with_super_admin_user), + ) + end = next(task for task in response.json if task["name"] == "End") + assert end["data"]["result"] == {"message": "message 1"} + + def test_manual_complete_task( + self, + app: Flask, + client: FlaskClient, + with_db_and_bpmn_file_cleanup: None, + with_super_admin_user: UserModel, + ) -> None: + """Test_script_unit_test_run.""" + process_group_id = "test_group" + process_model_id = "process_navigation" + bpmn_file_name = "process_navigation.bpmn" + bpmn_file_location = "process_navigation" + process_model_identifier = self.create_group_and_model_with_bpmn( + client=client, + user=with_super_admin_user, + process_group_id=process_group_id, + process_model_id=process_model_id, + bpmn_file_name=bpmn_file_name, + bpmn_file_location=bpmn_file_location, + ) + + bpmn_file_data_bytes = self.get_test_data_file_contents( + bpmn_file_name, bpmn_file_location + ) + self.create_spec_file( + client=client, + process_model_id=process_model_identifier, + process_model_location=process_model_identifier, + file_name=bpmn_file_name, + file_data=bpmn_file_data_bytes, + user=with_super_admin_user, + ) + + headers = self.logged_in_headers(with_super_admin_user) + response = self.create_process_instance_from_process_model_id_with_api( + client, process_model_identifier, headers + ) + process_instance_id = response.json["id"] + + client.post( + f"/v1.0/process-instances/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}/run", + headers=self.logged_in_headers(with_super_admin_user), + ) + + data = { + "dateTime": "timedelta(hours=1)", + "external": True, + "internal": True, + "label": "Event_0e4owa3", + "typename": "TimerEventDefinition", + } + response = client.post( + f"/v1.0/send-event/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}", + headers=self.logged_in_headers(with_super_admin_user), + content_type="application/json", + data=json.dumps(data), + ) + + response = client.get( + f"/v1.0/task-data/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}", + headers=self.logged_in_headers(with_super_admin_user), + ) + assert len(response.json) == 1 + task = response.json[0] + + response = client.post( + f"/v1.0/task-complete/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}/{task['id']}", + headers=self.logged_in_headers(with_super_admin_user), + content_type="application/json", + ) + assert response.json["status"] == "suspended" + def setup_initial_groups_for_move_tests( self, client: FlaskClient, with_super_admin_user: UserModel ) -> None: