sartography/spiffworkflow-backend
1
0
Fork 0
mirror of https://github.com/sartography/spiffworkflow-backend.git synced 2025-02-24 13:28:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

added api endpoint to return all tasks for a given process instance and updated task_show to return non-active task as well

Browse Source
This commit is contained in:
jasquat 2022-07-25 14:34:18 -04:00
parent f65228c87f
commit d355331cc2
4 changed files with 102 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/spiffworkflow_backend/api.yml
View File

@ -671,6 +671,26 @@ paths:
type: array
items:
$ref: "#/components/schemas/Task"
/process-instance/{process_instance_id}/tasks:
parameters:
- name: process_instance_id
in: path
required: true
description: The unique id of an existing process instance.
schema:
type: integer
get:
operationId: spiffworkflow_backend.routes.process_api_blueprint.process_instance_task_list
summary: returns the list of all user tasks associated with process instance
responses:
"200":
description: list of tasks
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/Task"
/tasks/{process_instance_id}/{task_id}:
parameters:
- name: task_id
@ -679,6 +699,12 @@ paths:
description: The unique id of an existing process group.
schema:
type: string
- name: process_instance_id
in: path
required: true
description: The unique id of an existing process instance.
schema:
type: integer
- name: terminate_loop
in: query
required: false

4
src/spiffworkflow_backend/models/active_task.py
View File

@ -26,10 +26,6 @@ class ActiveTaskModel(SpiffworkflowBaseDBModel):
),
)
# form_json: str | None = ""
# bpmn_json: str = ""
# preceding_spiffworkflow_user_task_id: int | None = None
assigned_principal: RelationshipProperty[PrincipalModel] = relationship(
PrincipalModel
)

6
src/spiffworkflow_backend/models/task.py
View File

@ -156,6 +156,10 @@ class Task:
@property
def serialized(self) -> dict[str, Any]:
"""Return object data in serializeable format."""
multi_instance_type = None
if self.multi_instance_type:
MultiInstanceType(self.multi_instance_type)
return {
"id": self.id,
"name": self.name,
@ -166,7 +170,7 @@ class Task:
"form": self.form,
"documentation": self.documentation,
"data": self.data,
"multi_instance_type": self.multi_instance_type,
"multi_instance_type": multi_instance_type,
"multi_instance_count": self.multi_instance_count,
"multi_instance_index": self.multi_instance_index,
"process_name": self.process_name,

180
src/spiffworkflow_backend/routes/process_api_blueprint.py
View File

@ -17,6 +17,7 @@ from flask.wrappers import Response
from flask_bpmn.api.api_error import ApiError
from flask_bpmn.models.db import db
from SpiffWorkflow import TaskState # type: ignore
from SpiffWorkflow import Task as SpiffTask # type: ignore
from sqlalchemy import desc
from spiffworkflow_backend.exceptions.process_entity_not_found_error import (
@ -554,23 +555,57 @@ def task_list_my_tasks(page: int = 1, per_page: int = 100) -> flask.wrappers.Res
return make_response(jsonify(response_json), 200)
def process_instance_task_list(process_instance_id: int) -> flask.wrappers.Response:
process_instance = find_process_instance_by_id_or_raise(
process_instance_id
)
processor = ProcessInstanceProcessor(process_instance)
all_spiff_user_tasks = processor.get_all_user_tasks()
tasks = [ProcessInstanceService.spiff_task_to_api_task(spiff_task) for spiff_task in all_spiff_user_tasks]
return make_response(jsonify(tasks), 200)
def task_show(process_instance_id: int, task_id: str) -> flask.wrappers.Response:
"""Task_list_my_tasks."""
principal = find_principal_or_raise()
active_task_assigned_to_me = find_active_task_by_id_or_raise(
process_instance_id, task_id, principal.id
)
process_instance = find_process_instance_by_id_or_raise(
active_task_assigned_to_me.process_instance_id
process_instance_id
)
process_model = get_process_model(
process_instance.process_model_identifier,
process_instance.process_group_identifier,
)
if active_task_assigned_to_me.form_file_name is None:
active_task_assigned_to_me = ActiveTaskModel.query.filter_by(
process_instance_id=process_instance_id,
task_id=task_id,
assigned_principal_id=principal.id,
).first()
form_schema_file_name = ""
form_ui_schema_file_name = ""
task = None
if active_task_assigned_to_me:
form_schema_file_name = active_task_assigned_to_me.form_file_name
form_ui_schema_file_name = active_task_assigned_to_me.ui_form_file_name
task = active_task_assigned_to_me.to_task()
else:
spiff_task = get_spiff_task_from_process_instance(task_id, process_instance)
extensions = spiff_task.task_spec.extensions
if "properties" in extensions:
properties = extensions["properties"]
if "formJsonSchemaFilename" in properties:
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)
task.data = spiff_task.data
if form_schema_file_name is None:
raise (
ApiError(
code="missing_form_file",
@ -580,19 +615,18 @@ def task_show(process_instance_id: int, task_id: str) -> flask.wrappers.Response
)
form_contents = prepare_form_data(
active_task_assigned_to_me.form_file_name,
json.loads(active_task_assigned_to_me.task_data),
form_schema_file_name,
task.data,
process_model,
)
task = active_task_assigned_to_me.to_task()
if form_contents:
task.form_schema = form_contents
if active_task_assigned_to_me.ui_form_file_name:
if form_ui_schema_file_name:
ui_form_contents = prepare_form_data(
active_task_assigned_to_me.ui_form_file_name,
json.loads(active_task_assigned_to_me.task_data),
form_ui_schema_file_name,
task.data,
process_model,
)
if ui_form_contents:
@ -601,89 +635,6 @@ def task_show(process_instance_id: int, task_id: str) -> flask.wrappers.Response
return make_response(jsonify(task), 200)
def task_show_completed_user_task(
process_instance_id: int, spiffworkflow_task_uuid: str
) -> flask.wrappers.Response:
"""Task_show_completed_user_task."""
process_instance = find_process_instance_by_id_or_raise(process_instance_id)
processor = ProcessInstanceProcessor(process_instance)
if processor.completed_user_tasks() is None:
raise (
ApiError(
code="no_completed_user_tasks",
message=f"This process instance does not have any completed user tasks: {process_instance_id}",
status_code=400,
)
)
spiffworkflow_task = None
preceding_spiffworkflow_user_task_id = None
following_spiffworkflow_user_task_id = None
for index, completed_user_task in enumerate(processor.completed_user_tasks()):
if str(completed_user_task.id) == spiffworkflow_task_uuid:
spiffworkflow_task = completed_user_task
preceding_spiffworkflow_user_task_id = get_value_from_array_with_index(
processor.completed_user_tasks(), index + 1
)
following_spiffworkflow_user_task_id = get_value_from_array_with_index(
processor.completed_user_tasks(), index - 1
)
if spiffworkflow_task is None:
raise (
ApiError(
code="no_completed_user_task_with_id",
message=f"This process instance does not have a completed user task with given id: {spiffworkflow_task_uuid}",
status_code=400,
)
)
process_model = get_process_model(
process_instance.process_model_identifier,
process_instance.process_group_identifier,
)
task = ProcessInstanceService.spiff_task_to_api_task(
spiffworkflow_task, add_docs_and_forms=True
)
form_contents = prepare_form_data(
task.properties["properties"]["formJsonSchemaFilename"],
spiffworkflow_task.data,
process_model,
)
if form_contents is not None:
task.form_schema = form_contents
task.data = spiffworkflow_task.data
if preceding_spiffworkflow_user_task_id:
task.preceding_spiffworkflow_user_task_id = (
preceding_spiffworkflow_user_task_id.id
)
if following_spiffworkflow_user_task_id:
task.following_spiffworkflow_user_task_id = (
following_spiffworkflow_user_task_id.id
)
principal = find_principal_or_raise()
active_task = (
ActiveTaskModel.query.filter_by(
assigned_principal_id=principal.id, process_instance_id=process_instance_id
).order_by(
desc(ActiveTaskModel.id) # type: ignore
)
).first()
task.current_active_task_id = active_task.id
return Response(
json.dumps(TaskSchema().dump(task)), status=200, mimetype="application/json"
)
def task_submit(
process_instance_id: int,
task_id: str,
@ -701,19 +652,8 @@ def task_submit(
)
processor = ProcessInstanceProcessor(process_instance)
task_uuid = uuid.UUID(
active_task_assigned_to_me.task_id
)
spiff_task = processor.bpmn_process_instance.get_task(task_uuid)
spiff_task = get_spiff_task_from_process_instance(task_id, process_instance, processor=processor)
if spiff_task is None:
raise (
ApiError(
code="empty_task",
message="Processor failed to obtain task.",
status_code=500,
)
)
if spiff_task.state != TaskState.READY:
raise (
ApiError(
@ -852,14 +792,36 @@ def get_value_from_array_with_index(array: list, index: int) -> Any:
def prepare_form_data(
form_file: str, spiffworkflow_data_json: dict, process_model: ProcessModelInfo
form_file: str, task_data: 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")
# trade out pieces like "{{variable_name}}" for the corresponding form data value
for key, value in spiffworkflow_data_json.items():
for key, value in task_data.items():
if isinstance(value, str) or isinstance(value, int):
file_contents = file_contents.replace("{{" + key + "}}", str(value))
return file_contents
def get_spiff_task_from_process_instance(task_id: str, process_instance: ProcessInstanceModel, processor: ProcessInstanceProcessor | None = None) -> SpiffTask:
if processor is None:
processor = ProcessInstanceProcessor(process_instance)
task_uuid = uuid.UUID(
task_id
)
spiff_task = processor.bpmn_process_instance.get_task(task_uuid)
if spiff_task is None:
raise (
ApiError(
code="empty_task",
message="Processor failed to obtain task.",
status_code=500,
)
)
return spiff_task