Task creation

Rule framework also allows you to create tasks from rules. This functionality can be used like any other function from the rule framework. This guide assumes you have already configured a taskboard. If not, please refer to this article: Task Management

The interface for the rule is as follows:

    def create_task(
        self,
        title: str,
        task_type: str,
        status: str,
        priority: TaskPriorityEnum,
        assignee_email: str,
        description: str = "",
        tags: list[str] | None = None,
        aggregation_configurations: list[TaskAggregationOptionEnum] | None = None,
        subject_datasource_id: str | None = None,
        subject_flow_config_id: str | None = None,
        ranking: int | None = None,
        taskboards: list[int] | None = None,
    ):
        """Creates a task from a rule.
        Args:
            title: The title of the task.
            task_type: The type of the task. It must be an existing Task Type configured in TaskManagement.
                It can be used for aggregating tasks if the aggregation_configuration is set to `task_type'.
            status: The status of the task, it must be the id of an existing TaskStatus configured in TaskManagement.
            priority: The priority of the task. It is of type TaskPriorityEnum.
            assignee_email: The assignee email of the task.
            description: The description of the task.
            tags: The tags of the task. It can be any list of strings.
            aggregation_configurations: The aggregation configuration of the task. It can be used to aggregate this
                task. If for example, `task_type` is in the list, the tasks will be aggregated by task type.
                It is a list of TaskAggregationOptionEnum.
            subject_datasource_id: The datasource id of the subject of the task. By default, it will be
                self.datasource.id.
            subject_flow_config_id: The flow configuration id of the subject of the task. By default, it will be
                self.flow_config.id.
            ranking: The ranking of the task, it can be any positive integer.
            taskboards: A list of the IDs of the taskboards that the task should be linked to
        """

Tasks created from the rules will be populated in the tasks management board.

Example of use:

     from energyworx_public.rule import TaskPriorityEnum, TaskAggregationOptionEnum
    
     def create_critical_event_task(self, error: str, assignee: str, taskboard: str | None = None):
            R""" Create a task for the critical event.
    
            Arguments:
                error: The error that the task is for
                assignee: The email of the user the task will be assigned to
                taskboard: The ID of the taskboard the task will be assigned to
    
            """
            title = f'Event occurred on datasource {self.datasource.id}'
            tags = ["meter_event"]
            aggregation_configurations = [TaskAggregationOptionEnum.datasource, TaskAggregationOptionEnum.task_type]
            priority = TaskPriorityEnum.medium
            status = StandardConstants.TASK_STATUS_NEW
            task_type = StandardConstants.TASK_TYPE_EVENT
            # taskboard needs to be of type list[int], so convert it to the correct format.
            if taskboard:
                try:
                    taskboard = [int(board) for board in taskboard.split(',')]
                except ValueError:
                    raise FlowCancelException(f"Parameter taskboard is not in the right format. Expected a comma-separated"
                                              f" list of integers, but got {taskboard}")
            description = (f'The flow {self.flow_id} for datasource '
                           f'{self.datasource.id} caused an error with code {error}.')
            self.create_task(
                title=title,
                description=description,
                tags=tags,
                aggregation_configurations=aggregation_configurations,
                priority=priority,
                status=status,
                task_type=task_type,
                assignee_email=assignee,
                taskboards=taskboard
            )

Limitations: as per 24.04 the following limitations apply:

• Status and Type Existence : The system doesn't check if the specified statuses and types are valid. If they don't exist, the task won't be created. There's currently no notification system to alert you about this.
• User Email : The system doesn't verify if the provided email address belongs to an existing user in the platform.