Merge pull request #101 from sartography/task_data_len

Throw an exception if max task data length is exceeded
This commit is contained in:
Kevin Burnett 2023-01-13 11:43:46 -08:00 committed by GitHub
commit 6fc41201b1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 26 additions and 0 deletions

View File

@ -1307,8 +1307,34 @@ class ProcessInstanceProcessor:
except WorkflowTaskExecException as we: except WorkflowTaskExecException as we:
raise ApiError.from_workflow_exception("task_error", str(we), we) from we raise ApiError.from_workflow_exception("task_error", str(we), we) from we
def user_defined_task_data(self, task_data: dict) -> dict:
"""UserDefinedTaskData."""
return {k: v for k, v in task_data.items() if k != "current_user"}
def check_task_data_size(self) -> None:
"""CheckTaskDataSize."""
tasks_to_check = self.bpmn_process_instance.get_tasks(TaskState.FINISHED_MASK)
task_data = [self.user_defined_task_data(task.data) for task in tasks_to_check]
task_data_to_check = list(filter(len, task_data))
try:
task_data_len = len(json.dumps(task_data_to_check))
except Exception:
task_data_len = 0
task_data_limit = 1024**2
if task_data_len > task_data_limit:
raise (
ApiError(
error_code="task_data_size_exceeded",
message=f"Maximum task data size of {task_data_limit} exceeded.",
)
)
def serialize(self) -> str: def serialize(self) -> str:
"""Serialize.""" """Serialize."""
self.check_task_data_size()
return self._serializer.serialize_json(self.bpmn_process_instance) # type: ignore return self._serializer.serialize_json(self.bpmn_process_instance) # type: ignore
def next_user_tasks(self) -> list[SpiffTask]: def next_user_tasks(self) -> list[SpiffTask]: