Skip to content

event

dev_tool.event

__all__ = ['Event', 'EventBus', 'EventEmitterMixin', 'EventHandler', 'emit', 'emit_error', 'emit_info', 'emit_normal', 'emit_success', 'emit_warning'] module-attribute

Event dataclass

A data class representing an event in the system.

This class contains all the information about an event including its type, data, timing, and tracking information.

type instance-attribute

data instance-attribute

timestamp instance-attribute

source instance-attribute

correlation_id instance-attribute

EventBus

A singleton event bus class for publishing and subscribing to events.

This class provides a centralized mechanism for event-driven communication throughout the application with thread-safe operations.

The constructor for the EventBus class.

Initializes the event bus with empty handlers and thread safety locks.

Source code in dev_tool/event/bus.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(self) -> None:
    """
    The constructor for the EventBus class.

    Initializes the event bus with empty handlers and thread safety locks.
    """

    if hasattr(self, '_initialized'):
        return

    self._handlers = {}
    self._initialized = True
    self.lock = threading.RLock()

lock = threading.RLock() instance-attribute

__new__

A class method that ensures singleton instance creation with thread safety.

Returns:

  • Self –

    The singleton EventBus instance.

Source code in dev_tool/event/bus.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def __new__(cls) -> Self:
    """
    A class method that ensures singleton instance creation with thread safety.

    :return: The singleton EventBus instance.
    """

    if cls._instance is None:
        with cls._lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)

    assert isinstance(cls._instance, cls)
    return cls._instance

emit

A method that emits an event to all registered handlers.

Parameters:

  • event_type (EventType) –

    The type of event to emit.

  • data (dict[str, str | int | bool]) –

    The event data dictionary.

  • source (str, default: 'unknown' ) –

    The source identifier for the event.

  • correlation_id (str | None, default: None ) –

    Optional correlation ID for event tracking.

Source code in dev_tool/event/bus.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def emit(
    self,
    event_type: EventType,
    data: dict[str, str | int | bool],
    source: str = 'unknown',
    correlation_id: str | None = None
) -> None:
    """
    A method that emits an event to all registered handlers.

    :param event_type: The type of event to emit.
    :param data: The event data dictionary.
    :param source: The source identifier for the event.
    :param correlation_id: Optional correlation ID for event tracking.
    """

    event = Event(
        type=event_type,
        data=data,
        timestamp=time.time(),
        source=source,
        correlation_id=correlation_id or str(uuid4())
    )

    self.publish(event)

publish

A method that publishes an event to all registered handlers.

Parameters:

  • event (Event) –

    The event object to publish to handlers.

Source code in dev_tool/event/bus.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
def publish(self, event: Event) -> None:
    """
    A method that publishes an event to all registered handlers.

    :param event: The event object to publish to handlers.
    """

    with self.lock:
        handlers = self._handlers.get(event.type, set())

    for handler in handlers.copy():
        with contextlib.suppress(Exception):
            handler.handle(event)

subscribe

A method that subscribes a handler to a specific event type.

Parameters:

  • event_type (EventType) –

    The type of event to subscribe to.

  • handler (EventHandler) –

    The event handler to register for the event type.

Source code in dev_tool/event/bus.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
def subscribe(self, event_type: EventType, handler: EventHandler) -> None:
    """
    A method that subscribes a handler to a specific event type.

    :param event_type: The type of event to subscribe to.
    :param handler: The event handler to register for the event type.
    """

    with self.lock:
        if event_type not in self._handlers:
            self._handlers[event_type] = set()

        self._handlers[event_type].add(handler)

unsubscribe

A method that unsubscribes a handler from a specific event type.

Parameters:

  • event_type (EventType) –

    The type of event to unsubscribe from.

  • handler (EventHandler) –

    The event handler to remove from the event type.

Source code in dev_tool/event/bus.py 
145
146
147
148
149
150
151
152
153
154
155
def unsubscribe(self, event_type: EventType, handler: EventHandler) -> None:
    """
    A method that unsubscribes a handler from a specific event type.

    :param event_type: The type of event to unsubscribe from.
    :param handler: The event handler to remove from the event type.
    """

    with self.lock:
        if event_type in self._handlers:
            self._handlers[event_type].discard(handler)

EventHandler

Bases: Protocol

A protocol for event handlers.

This class defines the interface that all event handlers must implement to process events from the event bus.

handle

A method that handles an event.

Parameters:

  • event (Event) –

    The event object to handle.

Source code in dev_tool/event/bus.py 
41
42
43
44
45
46
47
48
def handle(self, event: Event) -> None:
    """
    A method that handles an event.

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

    ...

EventEmitterMixin

event_manager property

emit

Source code in dev_tool/event/emit.py 
49
50
51
def emit(self, event_type: AppEventType, data: dict | None = None, source: str | None = None) -> None:
    event_source = source or self.__class__.__name__
    self.event_manager.emit(event_type, data, source=event_source)

emit_error

Source code in dev_tool/event/emit.py 
53
54
55
56
57
def emit_error(self, message: str, stacktrace: str | None = None) -> None:
    self.emit(
        AppEventType.NOTIFICATION_ERROR_BANNER,
        {'message': message, 'stacktrace': stacktrace}
    )

emit_info

Source code in dev_tool/event/emit.py 
59
60
def emit_info(self, message: str) -> None:
    self.emit(AppEventType.NOTIFICATION_INFO, {'message': message})

emit_normal

Source code in dev_tool/event/emit.py 
62
63
def emit_normal(self, message: str) -> None:
    self.emit(AppEventType.NOTIFICATION_NORMAL, {'message': message})

emit_success

Source code in dev_tool/event/emit.py 
65
66
def emit_success(self, message: str) -> None:
    self.emit(AppEventType.NOTIFICATION_SUCCESS_BANNER, {'message': message})

emit_warning

Source code in dev_tool/event/emit.py 
68
69
def emit_warning(self, message: str) -> None:
    self.emit(AppEventType.NOTIFICATION_WARNING, {'message': message})

emit_error

Source code in dev_tool/event/emit.py 
20
21
22
23
24
25
def emit_error(message: str, stacktrace: str | None = None, source: str = 'app') -> None:
    emit(
        AppEventType.NOTIFICATION_ERROR_BANNER,
        {'message': message, 'stacktrace': stacktrace},
        source=source
    )

emit_info

Source code in dev_tool/event/emit.py 
28
29
def emit_info(message: str, source: str = 'app') -> None:
    emit(AppEventType.NOTIFICATION_INFO, {'message': message}, source=source)

emit_normal

Source code in dev_tool/event/emit.py 
32
33
def emit_normal(message: str, source: str = 'app') -> None:
    emit(AppEventType.NOTIFICATION_NORMAL, {'message': message}, source=source)

emit_success

Source code in dev_tool/event/emit.py 
36
37
def emit_success(message: str, source: str = 'app') -> None:
    emit(AppEventType.NOTIFICATION_SUCCESS_BANNER, {'message': message}, source=source)

emit_warning

Source code in dev_tool/event/emit.py 
40
41
def emit_warning(message: str, source: str = 'app') -> None:
    emit(AppEventType.NOTIFICATION_WARNING, {'message': message}, source=source)