run_pyl

spiffworkflow-backend/src/spiffworkflow_backend/config/__init__.py

@@ -18,13 +18,13 @@ def setup_database_uri(app: Flask) -> None:
     if app.config.get("SPIFFWORKFLOW_BACKEND_DATABASE_URI") is None:
         database_name = f"spiffworkflow_backend_{app.config['ENV_IDENTIFIER']}"
         if app.config.get("SPIFFWORKFLOW_BACKEND_DATABASE_TYPE") == "sqlite":
-            app.config["SQLALCHEMY_DATABASE_URI"] = (
-                f"sqlite:///{app.instance_path}/db_{app.config['ENV_IDENTIFIER']}.sqlite3"
-            )
+            app.config[
+                "SQLALCHEMY_DATABASE_URI"
+            ] = f"sqlite:///{app.instance_path}/db_{app.config['ENV_IDENTIFIER']}.sqlite3"
         elif app.config.get("SPIFFWORKFLOW_BACKEND_DATABASE_TYPE") == "postgres":
-            app.config["SQLALCHEMY_DATABASE_URI"] = (
-                f"postgresql://spiffworkflow_backend:spiffworkflow_backend@localhost:5432/{database_name}"
-            )
+            app.config[
+                "SQLALCHEMY_DATABASE_URI"
+            ] = f"postgresql://spiffworkflow_backend:spiffworkflow_backend@localhost:5432/{database_name}"
         else:
             # use pswd to trick flake8 with hardcoded passwords
             db_pswd = app.config.get("SPIFFWORKFLOW_BACKEND_DATABASE_PASSWORD")

spiffworkflow-backend/src/spiffworkflow_backend/models/process_instance.py

@@ -129,9 +129,9 @@ class ProcessInstanceModel(SpiffworkflowBaseDBModel):
     def serialized_with_metadata(self) -> dict[str, Any]:
         process_instance_attributes = self.serialized
         process_instance_attributes["process_metadata"] = self.process_metadata
-        process_instance_attributes["process_model_with_diagram_identifier"] = (
-            self.process_model_with_diagram_identifier
-        )
+        process_instance_attributes[
+            "process_model_with_diagram_identifier"
+        ] = self.process_model_with_diagram_identifier
         return process_instance_attributes
 
     @property

spiffworkflow-backend/src/spiffworkflow_backend/services/process_instance_processor.py

@@ -423,9 +423,9 @@ class ProcessInstanceProcessor:
         tld.process_instance_id = process_instance_model.id
 
         # we want this to be the fully qualified path to the process model including all group subcomponents
-        current_app.config["THREAD_LOCAL_DATA"].process_model_identifier = (
-            f"{process_instance_model.process_model_identifier}"
-        )
+        current_app.config[
+            "THREAD_LOCAL_DATA"
+        ].process_model_identifier = f"{process_instance_model.process_model_identifier}"
 
         self.process_instance_model = process_instance_model
         self.process_model_service = ProcessModelService()
@@ -585,9 +585,9 @@ class ProcessInstanceProcessor:
                 bpmn_subprocess_definition.bpmn_identifier
             ] = bpmn_process_definition_dict
             spiff_bpmn_process_dict["subprocess_specs"][bpmn_subprocess_definition.bpmn_identifier]["task_specs"] = {}
-            bpmn_subprocess_definition_bpmn_identifiers[bpmn_subprocess_definition.id] = (
-                bpmn_subprocess_definition.bpmn_identifier
-            )
+            bpmn_subprocess_definition_bpmn_identifiers[
+                bpmn_subprocess_definition.id
+            ] = bpmn_subprocess_definition.bpmn_identifier
 
         task_definitions = TaskDefinitionModel.query.filter(
             TaskDefinitionModel.bpmn_process_definition_id.in_(  # type: ignore
@@ -1211,11 +1211,12 @@ class ProcessInstanceProcessor:
             raise TaskNotFoundError(
                 f"Cannot find a task with guid '{to_task_guid}' for process instance '{process_instance.id}'"
             )
-        # If this task model has a parent boundary event, reset to that point instead, so we can reset all the boundary timers, etc...
-        parent_id = to_task_model.properties_json.get('parent','')
+        # If this task model has a parent boundary event, reset to that point instead,
+        # so we can reset all the boundary timers, etc...
+        parent_id = to_task_model.properties_json.get("parent", "")
         parent = TaskModel.query.filter_by(guid=parent_id).first()
         is_boundary_parent = False
-        if parent and parent.task_definition.typename == '_BoundaryEventParent':
+        if parent and parent.task_definition.typename == "_BoundaryEventParent":
             to_task_model = parent
             is_boundary_parent = True  # Will need to complete this task at the end so we are on the correct process.
 
@@ -1325,7 +1326,7 @@ class ProcessInstanceProcessor:
 
         # If this as a boundary event parent, run it, so we get back to an active task.
         if is_boundary_parent:
-            processor.do_engine_steps(execution_strategy_name='one_at_a_time')
+            processor.do_engine_steps(execution_strategy_name="one_at_a_time")
 
         processor.save()
         processor.suspend()

spiffworkflow-backend/tests/spiffworkflow_backend/unit/test_process_instance_processor.py

@@ -454,21 +454,23 @@ class TestProcessInstanceProcessor(BaseTest):
         assert len(process_instance.active_human_tasks) == 1
         human_task_one = process_instance.active_human_tasks[0]
         spiff_manual_task = processor.bpmn_process_instance.get_task_from_id(UUID(human_task_one.task_id))
-        ProcessInstanceService.complete_form_task(processor, spiff_manual_task, {}, with_super_admin_user, human_task_one)
+        ProcessInstanceService.complete_form_task(
+            processor, spiff_manual_task, {}, with_super_admin_user, human_task_one
+        )
         assert (
             len(process_instance.active_human_tasks) == 1
         ), "expected 1 active human tasks after 2nd one is completed"
-        assert process_instance.active_human_tasks[0].task_title == 'Final'
+        assert process_instance.active_human_tasks[0].task_title == "Final"
 
         # Reset the process back to the task within the call activity that contains a timer_boundary event.
-        reset_to_spiff_task = processor.__class__.get_task_by_bpmn_identifier(
-            'manual_task_1', processor.bpmn_process_instance
+        reset_to_spiff_task: SpiffTask = processor.__class__.get_task_by_bpmn_identifier(
+            "manual_task_1", processor.bpmn_process_instance
         )
         processor.suspend()
         processor = ProcessInstanceProcessor(process_instance)
         ProcessInstanceProcessor.reset_process(process_instance, str(reset_to_spiff_task.id))
         human_task_one = process_instance.active_human_tasks[0]
-        assert human_task_one.task_title == 'Manual Task #1'
+        assert human_task_one.task_title == "Manual Task #1"
         processor = ProcessInstanceProcessor(process_instance)
         processor.manual_complete_task(str(spiff_manual_task.id), execute=True)
         processor = ProcessInstanceProcessor(process_instance)
@@ -476,10 +478,11 @@ class TestProcessInstanceProcessor(BaseTest):
         processor.do_engine_steps(save=True)
         process_instance = ProcessInstanceModel.query.filter_by(id=process_instance.id).first()
 
-        assert (len(process_instance.active_human_tasks) == 1)
-        assert process_instance.active_human_tasks[0].task_title == 'Final', \
-            "once we reset, resume, and complete the task, we should be back to the Final step again, and not" \
+        assert len(process_instance.active_human_tasks) == 1
+        assert process_instance.active_human_tasks[0].task_title == "Final", (
+            "once we reset, resume, and complete the task, we should be back to the Final step again, and not"
             "stuck waiting for the call activity to complete (which was happening in a bug I'm fixing right now)"
+        )
 
     def test_properly_saves_tasks_when_running(
         self,