task

dev_tool.event.handler.task

log = logging.getLogger(__name__) module-attribute

TaskEventHandler

A handler class for task lifecycle events.

This class manages watchers for task events and notifies them when task state changes occur.

The constructor for the TaskEventHandler class.

Initializes an empty set of watchers.

Source code in dev_tool/event/handler/task.py

def __init__(self) -> None:
    """
    The constructor for the TaskEventHandler class.

    Initializes an empty set of watchers.
    """

    self._watchers: set[Callable[[dict], None]] = set()

handle

A method that handles task lifecycle events.

Parameters:

• event (Event) –

  The event object to handle.

Source code in dev_tool/event/handler/task.py

def handle(self, event: Event) -> None:
    """
    A method that handles task lifecycle events.

    :param event: The event object to handle.
    """

    for watcher in self._watchers.copy():
        try:
            watcher(dict(event.data))
        except Exception:
            log.exception(f'Task event watcher failed for {event.type}')

register_watcher

A method that registers a watcher for task events.

Parameters:

• watcher (Callable[[dict], None]) –

  The watcher function to register.

Source code in dev_tool/event/handler/task.py

def register_watcher(self, watcher: Callable[[dict], None]) -> None:
    """
    A method that registers a watcher for task events.

    :param watcher: The watcher function to register.
    """

    self._watchers.add(watcher)

unregister_watcher

A method that unregisters a watcher for task events.

Parameters:

• watcher (Callable[[dict], None]) –

  The watcher function to unregister.

Source code in dev_tool/event/handler/task.py

def unregister_watcher(self, watcher: Callable[[dict], None]) -> None:
    """
    A method that unregisters a watcher for task events.

    :param watcher: The watcher function to unregister.
    """

    self._watchers.discard(watcher)

get_event_type staticmethod

A method that returns the event types this handler processes.

Returns:

• list[AppEventType] –

  A list of event types.

Source code in dev_tool/event/handler/task.py

@staticmethod
def get_event_type() -> list[AppEventType]:
    """
    A method that returns the event types this handler processes.

    :return: A list of event types.
    """

    return [
        AppEventType.TASK_CANCELLED,
        AppEventType.TASK_COMPLETED,
        AppEventType.TASK_FAILED,
        AppEventType.TASK_PROGRESS,
        AppEventType.TASK_STARTED,
    ]