Method set

SpiffWorkflow/bpmn/specs/data_spec.py:78–91

Copy a value from the task data to the workflow data

(self, my_task)

        logger.info(f'Read workflow variable', extra=my_task.collect_log_extras({'bpmn_id': self.bpmn_id}))

    def set(self, my_task):
        """Copy a value from the task data to the workflow data"""

        if self.bpmn_id not in my_task.data:
            message = f"A data object could not be set; '{self.bpmn_id}' not exist in the task."
            raise WorkflowDataException(message, my_task, data_output=self)

        wf = my_task.workflow
        while wf is not None and self.bpmn_id not in wf.spec.data_objects:
            wf = wf.parent_workflow

        wf.data_objects[self.bpmn_id] = deepcopy(my_task.data[self.bpmn_id])
        del my_task.data[self.bpmn_id]
        logger.info(f'Set workflow variable', extra=my_task.collect_log_extras({'bpmn_id': self.bpmn_id}))

    def delete(self, my_task):
        my_task.data.pop(self.bpmn_id, None)