Skip to content

handler

dev_tool.event.handler

__all__ = ['ConfigEventHandler', 'FileEventHandler', 'NotificationEventHandler', 'TaskEventHandler'] module-attribute

ConfigEventHandler

A handler class for configuration change events.

This class manages callbacks for configuration changes and executes them when corresponding events are received.

The constructor for the ConfigEventHandler class.

Initializes empty callback lists for each configuration event type.

Source code in dev_tool/event/handler/config.py 
26
27
28
29
30
31
32
33
34
35
36
def __init__(self) -> None:
    """
    The constructor for the ConfigEventHandler class.

    Initializes empty callback lists for each configuration event type.
    """

    self._callbacks: dict[AppEventType, list[Callable[[], None]]] = {
        AppEventType.CONFIG_ENV_CHANGED: [],
        AppEventType.CONFIG_PYPROJECT_CHANGED: [],
    }

handle

A method that handles configuration change events.

Parameters:

  • event (Event) –

    The event object to handle.

Source code in dev_tool/event/handler/config.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def handle(self, event: Event) -> None:
    """
    A method that handles configuration change events.

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

    event_type = cast('AppEventType', event.type)
    callbacks = self._callbacks.get(event_type, [])

    for callback in callbacks:
        try:
            callback()
        except Exception:
            log.exception(f'Config event callback failed for {event.type}')

register_callback

A method that registers a callback for a configuration event.

Parameters:

  • event_type (AppEventType) –

    The event type to listen for.

  • callback (Callable[[], None]) –

    The callback function to execute.

Source code in dev_tool/event/handler/config.py 
54
55
56
57
58
59
60
61
62
63
def register_callback(self, event_type: AppEventType, callback: Callable[[], None]) -> None:
    """
    A method that registers a callback for a configuration event.

    :param event_type: The event type to listen for.
    :param callback: The callback function to execute.
    """

    if event_type in self._callbacks:
        self._callbacks[event_type].append(callback)

unregister_callback

A method that unregisters a callback for a configuration event.

Parameters:

  • event_type (AppEventType) –

    The event type to stop listening for.

  • callback (Callable[[], None]) –

    The callback function to remove.

Source code in dev_tool/event/handler/config.py 
65
66
67
68
69
70
71
72
73
74
def unregister_callback(self, event_type: AppEventType, callback: Callable[[], None]) -> None:
    """
    A method that unregisters a callback for a configuration event.

    :param event_type: The event type to stop listening for.
    :param callback: The callback function to remove.
    """

    if event_type in self._callbacks and callback in self._callbacks[event_type]:
        self._callbacks[event_type].remove(callback)

get_event_type staticmethod

A method that returns the event types this handler processes.

Returns:

Source code in dev_tool/event/handler/config.py 
76
77
78
79
80
81
82
83
84
85
86
87
@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.CONFIG_ENV_CHANGED,
        AppEventType.CONFIG_PYPROJECT_CHANGED,
    ]

FileEventHandler

A handler class for file system events.

This class manages callbacks for file changes and executes them when corresponding events are received.

The constructor for the FileEventHandler class.

Initializes empty callback dictionaries for file events.

Source code in dev_tool/event/handler/file.py 
27
28
29
30
31
32
33
34
def __init__(self) -> None:
    """
    The constructor for the FileEventHandler class.

    Initializes empty callback dictionaries for file events.
    """

    self._callbacks: dict[str, list[Callable[[Path], None]]] = {}

handle

A method that handles file system events.

Parameters:

  • event (Event) –

    The event object to handle.

Source code in dev_tool/event/handler/file.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def handle(self, event: Event) -> None:
    """
    A method that handles file system events.

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

    path_value = event.data.get('path', '')

    if not path_value or not isinstance(path_value, str):
        return

    path_str: str = path_value
    path = Path(path_str)

    callbacks = self._callbacks.get(path_str, [])

    for callback in callbacks:
        try:
            callback(path)
        except Exception:
            log.exception(f'File event callback failed for {path_str}')

register_callback

A method that registers a callback for file changes.

Parameters:

  • path (Path) –

    The file path to watch.

  • callback (Callable[[Path], None]) –

    The callback function to execute.

Source code in dev_tool/event/handler/file.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def register_callback(self, path: Path, callback: Callable[[Path], None]) -> None:
    """
    A method that registers a callback for file changes.

    :param path: The file path to watch.
    :param callback: The callback function to execute.
    """

    path_str = str(path.absolute())

    if path_str not in self._callbacks:
        self._callbacks[path_str] = []

    if callback not in self._callbacks[path_str]:
        self._callbacks[path_str].append(callback)

unregister_callback

A method that unregisters a callback for file changes.

Parameters:

  • path (Path) –

    The file path to stop watching.

  • callback (Callable[[Path], None] | None, default: None ) –

    The specific callback to remove, or None to remove all.

Source code in dev_tool/event/handler/file.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def unregister_callback(self, path: Path, callback: Callable[[Path], None] | None = None) -> None:
    """
    A method that unregisters a callback for file changes.

    :param path: The file path to stop watching.
    :param callback: The specific callback to remove, or None to remove all.
    """

    path_str = str(path.absolute())

    if path_str not in self._callbacks:
        return

    if callback is None:
        del self._callbacks[path_str]
        return

    if callback in self._callbacks[path_str]:
        self._callbacks[path_str].remove(callback)

        if not self._callbacks[path_str]:
            del self._callbacks[path_str]

get_event_type staticmethod

A method that returns the event types this handler processes.

Returns:

Source code in dev_tool/event/handler/file.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@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.FILE_CHANGED,
        AppEventType.FILE_CREATED,
        AppEventType.FILE_DELETED,
    ]

NotificationEventHandler

A handler class for notification events.

This class processes notification events and delegates them to the notification service for display.

The constructor for the NotificationEventHandler class.

Parameters:

Source code in dev_tool/event/handler/notification.py 
20
21
22
23
24
25
26
27
def __init__(self, notification_service: NotificationService) -> None:
    """
    The constructor for the NotificationEventHandler class.

    :param notification_service: The notification service instance.
    """

    self.notification_service = notification_service

notification_service = notification_service instance-attribute

handle

A method that handles notification events.

Parameters:

  • event (Event) –

    The event object to handle.

Source code in dev_tool/event/handler/notification.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def handle(self, event: Event) -> None:
    """
    A method that handles notification events.

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

    message = str(event.data.get('message', ''))
    duration_value = event.data.get('duration', 5.0)
    persistent_value = event.data.get('persistent', False)
    identifier_value = event.data.get('identifier')
    stacktrace_value = event.data.get('stacktrace')

    duration = float(duration_value) if duration_value is not None else 5.0
    persistent = bool(persistent_value)
    identifier = str(identifier_value) if identifier_value is not None else None
    stacktrace = str(stacktrace_value) if stacktrace_value is not None else None

    match event.type:
        case AppEventType.NOTIFICATION_ERROR:
            self.notification_service.error_text(message)
        case AppEventType.NOTIFICATION_ERROR_BANNER:
            self.notification_service.error_banner(
                message,
                duration=duration,
                persistent=persistent,
                identifier=identifier,
                stacktrace=stacktrace
            )
        case AppEventType.NOTIFICATION_INFO:
            self.notification_service.info(message)
        case AppEventType.NOTIFICATION_INFO_BANNER:
            self.notification_service.info_banner(
                message,
                duration=duration,
                persistent=persistent,
                identifier=identifier
            )
        case AppEventType.NOTIFICATION_NORMAL:
            self.notification_service.normal_text(message)
        case AppEventType.NOTIFICATION_SUCCESS:
            self.notification_service.success_text(message)
        case AppEventType.NOTIFICATION_SUCCESS_BANNER:
            self.notification_service.success_banner(
                message,
                duration=duration,
                persistent=persistent,
                identifier=identifier
            )
        case AppEventType.NOTIFICATION_WARNING:
            self.notification_service.warning_text(message)

get_event_type staticmethod

A method that returns the event types this handler processes.

Returns:

Source code in dev_tool/event/handler/notification.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@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.NOTIFICATION_ERROR,
        AppEventType.NOTIFICATION_ERROR_BANNER,
        AppEventType.NOTIFICATION_INFO,
        AppEventType.NOTIFICATION_INFO_BANNER,
        AppEventType.NOTIFICATION_NORMAL,
        AppEventType.NOTIFICATION_SUCCESS,
        AppEventType.NOTIFICATION_SUCCESS_BANNER,
        AppEventType.NOTIFICATION_WARNING,
    ]

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 
26
27
28
29
30
31
32
33
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 
35
36
37
38
39
40
41
42
43
44
45
46
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 
48
49
50
51
52
53
54
55
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 
57
58
59
60
61
62
63
64
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:

Source code in dev_tool/event/handler/task.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@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,
    ]