import uuid
from typing import Any
from collections.abc import Callable, Iterable
from ppc_robot_lib.steps import abstract_step
from ppc_robot_lib.tasks.types import TaskLevel
ReportFunc = Callable[[dict[str, Any]], None]
[docs]
class Task:
"""
Represents a currently executing task (instance of job).
"""
def __init__(
self,
task_id: uuid.UUID,
task_level: TaskLevel,
steps: Iterable['abstract_step.AbstractStep'] | None = None,
report_function: ReportFunc | None = None,
parameters: dict[str, Any] = None,
):
self._id = task_id
self._task_level = task_level
self._steps = steps
self._report_function = report_function
self._parameters = parameters
@property
def id(self) -> uuid.UUID:
"""
:return: Task ID.
"""
return self._id
@property
def task_level(self) -> TaskLevel:
return self._task_level
def is_service_account_level(self):
return self._task_level == TaskLevel.SERVICE_ACCOUNT
def is_client_account_level(self):
return self._task_level == TaskLevel.CLIENT_ACCOUNT
def is_client_level(self):
return self._task_level == TaskLevel.CLIENT
@property
def uses_function_execution_model(self):
"""
:return: Whether the task should use the new function execution model.
"""
return self._report_function is not None
@property
def steps(self) -> Iterable['abstract_step.AbstractStep']:
"""
:return: List of task steps.
"""
return self._steps
@property
def report_function(self) -> ReportFunc:
"""
:return: Function that should be called in order to run the report.
"""
return self._report_function
@property
def parameters(self) -> dict[str, Any]:
"""
:return: Report parameters that should be passed to the :py:attr:`report_function`.
"""
return self._parameters