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 ewx_public.enums 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
)
Updating and reading tasks
A rule can also look up existing tasks and change them.
get_tasks returns the namespace's tasks that match every filter you pass, at most 100:
def get_tasks(
self,
subject_datasource_id: str | None = None, # tasks about this datasource
status: str | None = None, # status name, case-insensitive
task_type: str | None = None, # task type name, case-insensitive
tags: list[str] | None = None, # tasks carrying all of these tags
open_only: bool = True, # leave out archived tasks
) -> list[dict]
Each task is a dict with id, unique_identifier, title, description, task_type, status, priority (low, medium or high), assignee_email, tags and archived. Deleted tasks are never returned. The result is cached for the rest of the message, so asking twice costs one lookup.
update_task changes an existing task, named by exactly one of task_id or unique_identifier (as returned by get_tasks). Only the arguments you pass change:
def update_task(
self,
task_id: str | None = None,
unique_identifier: str | None = None,
status: str | None = None, # an existing task status, by name
priority: TaskPriority | str | None = None,
assignee_email: str | None = None, # "" unassigns the task
tags: list[str] | None = None, # replaces the current tags
description: str | None = None,
) -> None
The change shows up in the task's history with the flow as its author (flowId:<flow id>).
Example of use:
def apply(self, **kwargs):
# Close the open VEE tasks of this datasource once its data is valid again.
for task in self.get_tasks(subject_datasource_id=self.datasource.id, task_type="VEE", status="Open"):
self.update_task(task_id=task["id"], status="Closed", priority="low")
Things to keep in mind:
- Updates are applied after the flow run. A
get_taskslater in the same run still returns the old values, and a task created bycreate_taskin the same run cannot be updated in that run. - Missing tasks are skipped. If the task does not exist, was deleted or is read-only, the update is skipped and a warning audit event is written; the flow does not fail. An unknown status name does fail the update, as it does for
create_task. - In the local test framework, seed tasks with
rule.backend.add_task({...})(usesubject_datasource_idfor the datasource) and checkrule.backend.task_updatesfor the updates a rule made.
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.