event

dev_tool.event

__all__ = ['Event', 'EventBus', 'EventHandler'] 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

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

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)

    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

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

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

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

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

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

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

    ...