app

dev_tool.event.app

__all__ = ['AppEventManager', 'AppEventType', 'AppLifecycleHandler'] module-attribute

AppEventType

Bases: StrEnum

APP_CLOSED = 'app_closed' class-attribute instance-attribute

APP_CLOSING = 'app_closing' class-attribute instance-attribute

APP_STARTED = 'app_started' class-attribute instance-attribute

APP_STARTING = 'app_starting' class-attribute instance-attribute

COMMAND_COMPLETED = 'command_completed' class-attribute instance-attribute

COMMAND_FAILURE = 'command_failure' class-attribute instance-attribute

COMMAND_STARTING = 'command_starting' class-attribute instance-attribute

COMMAND_SUCCESS = 'command_success' class-attribute instance-attribute

CONFIG_ENV_CHANGED = 'config_env_changed' class-attribute instance-attribute

CONFIG_PYPROJECT_CHANGED = 'config_pyproject_changed' class-attribute instance-attribute

DATABASE_DUMP_COMPLETED = 'database_dump_completed' class-attribute instance-attribute

DATABASE_DUMP_STARTED = 'database_dump_started' class-attribute instance-attribute

DATABASE_MIGRATED = 'database_migrated' class-attribute instance-attribute

DATABASE_MIGRATING = 'database_migrating' class-attribute instance-attribute

DATABASE_SEEDED = 'database_seeded' class-attribute instance-attribute

DATABASE_SEEDING = 'database_seeding' class-attribute instance-attribute

DOCKER_CONTAINER_FAILED = 'docker_container_failed' class-attribute instance-attribute

DOCKER_CONTAINER_STARTED = 'docker_container_started' class-attribute instance-attribute

DOCKER_CONTAINER_STOPPED = 'docker_container_stopped' class-attribute instance-attribute

DOCKER_IMAGE_BUILT = 'docker_image_built' class-attribute instance-attribute

FILE_CHANGED = 'file_changed' class-attribute instance-attribute

FILE_CREATED = 'file_created' class-attribute instance-attribute

FILE_DELETED = 'file_deleted' class-attribute instance-attribute

NOTIFICATION_ERROR = 'notification_error' class-attribute instance-attribute

NOTIFICATION_ERROR_BANNER = 'notification_error_banner' class-attribute instance-attribute

NOTIFICATION_INFO = 'notification_info' class-attribute instance-attribute

NOTIFICATION_INFO_BANNER = 'notification_info_banner' class-attribute instance-attribute

NOTIFICATION_NORMAL = 'notification_normal' class-attribute instance-attribute

NOTIFICATION_SUCCESS = 'notification_success' class-attribute instance-attribute

NOTIFICATION_SUCCESS_BANNER = 'notification_success_banner' class-attribute instance-attribute

NOTIFICATION_WARNING = 'notification_warning' class-attribute instance-attribute

SERVICE_FAILED = 'service_failed' class-attribute instance-attribute

SERVICE_INITIALIZED = 'service_initialized' class-attribute instance-attribute

TASK_CANCELLED = 'task_cancelled' class-attribute instance-attribute

TASK_COMPLETED = 'task_completed' class-attribute instance-attribute

TASK_FAILED = 'task_failed' class-attribute instance-attribute

TASK_PROGRESS = 'task_progress' class-attribute instance-attribute

TASK_STARTED = 'task_started' class-attribute instance-attribute

AppLifecycleHandler

Bases: EventHandler

A handler class for application lifecycle events.

This class manages hooks for different application events and executes them when corresponding events are received.

The constructor for the AppLifecycleHandler class.

Initializes an empty hooks dictionary for storing event callbacks.

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

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

    Initializes an empty hooks dictionary for storing event callbacks.
    """

    self.hooks = {}

hooks = {} instance-attribute

handle

A method that handles incoming events by executing registered hooks.

Parameters:

• event (Event) –

  The event object to handle.

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

def handle(self, event: Event) -> None:
    """
    A method that handles incoming events by executing registered hooks.

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

    if event.type in self.hooks:
        for hook in self.hooks[event.type]:
            with contextlib.suppress(Exception):
                hook(event)

register_hook

A method that registers a hook function for a specific event type.

Parameters:

• event_type (AppEventType) –

  The application event type to register the hook for.

• function (Callable) –

  The callback function to execute when the event occurs.

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

def register_hook(self, event_type: AppEventType, function: Callable) -> None:
    """
    A method that registers a hook function for a specific event type.

    :param event_type: The application event type to register the hook for.
    :param function: The callback function to execute when the event occurs.
    """

    if event_type not in self.hooks:
        self.hooks[event_type] = []

    self.hooks[event_type].append(function)

AppEventManager

A manager class for application lifecycle events.

This class handles the registration and emission of application events through an event bus system with lifecycle hooks.

The constructor for the AppEventManager class.

Parameters:

• event_bus (EventBus) –

  The event bus instance for publishing events.

Source code in dev_tool/event/app/manager.py

def __init__(self, event_bus: EventBus) -> None:
    """
    The constructor for the AppEventManager class.

    :param event_bus: The event bus instance for publishing events.
    """

    self._notification_handler: NotificationEventHandler | None = None

    self.config_handler = ConfigEventHandler()
    self.event_bus = event_bus
    self.file_handler = FileEventHandler()
    self.handler = AppLifecycleHandler()
    self.task_handler = TaskEventHandler()

    for event_type in cast('Iterable[AppEventType]', AppEventType):
        self.event_bus.subscribe(event_type, self.handler)

    self._register_domain_handlers()

config_handler = ConfigEventHandler() instance-attribute

event_bus = event_bus instance-attribute

file_handler = FileEventHandler() instance-attribute

handler = AppLifecycleHandler() instance-attribute

task_handler = TaskEventHandler() instance-attribute

emit

A method that emits an application event.

Parameters:

• event_type (AppEventType) –

  The type of event to emit.

• data (dict[str, Any] | None, default: None ) –

  Optional event data dictionary.

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

  The source of the event.

Source code in dev_tool/event/app/manager.py

def emit(self, event_type: AppEventType, data: dict[str, Any] | None = None, source: str = 'app') -> None:
    """
    A method that emits an application event.

    :param event_type: The type of event to emit.
    :param data: Optional event data dictionary.
    :param source: The source of the event.
    """

    self.event_bus.emit(event_type, data or {}, source=source)

emit_app_event

A method that emits an application lifecycle event.

Parameters:

• event_type (AppEventType) –

  The type of application event to emit.

• data (dict[str, Any] | None, default: None ) –

  Optional event data dictionary.

Source code in dev_tool/event/app/manager.py

def emit_app_event(self, event_type: AppEventType, data: dict[str, Any] | None = None) -> None:
    """
    A method that emits an application lifecycle event.

    :param event_type: The type of application event to emit.
    :param data: Optional event data dictionary.
    """

    self.event_bus.emit(event_type, data or {}, source='app')

initialize_notification_handler

A method that initializes the notification handler.

Parameters:

• notification_service (NotificationService) –

  The notification service instance.

Source code in dev_tool/event/app/manager.py

def initialize_notification_handler(self, notification_service: NotificationService) -> None:
    """
    A method that initializes the notification handler.

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

    self._notification_handler = NotificationEventHandler(notification_service)

    for event_type in NotificationEventHandler.get_event_type():
        self.event_bus.subscribe(event_type, self._notification_handler)

register_config_callback

A method that registers a callback for configuration events.

Parameters:

• event_type (AppEventType) –

  The configuration event type to listen for.

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

  The callback function to execute.

Source code in dev_tool/event/app/manager.py

def register_config_callback(self, event_type: AppEventType, callback: Callable[[], None]) -> None:
    """
    A method that registers a callback for configuration events.

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

    self.config_handler.register_callback(event_type, callback)

register_file_callback

A method that registers a callback for file events.

Parameters:

• path (Path) –

  The file path to watch.

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

  The callback function to execute.

Source code in dev_tool/event/app/manager.py

def register_file_callback(self, path: Path, callback: Callable[[Path], None]) -> None:
    """
    A method that registers a callback for file events.

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

    self.file_handler.register_callback(path, callback)

register_hook

A method that registers a function to run when an application event occurs.

Parameters:

• event_type (AppEventType) –

  The application event type to listen for.

• function (Callable) –

  The callback function to execute when the event occurs.

Source code in dev_tool/event/app/manager.py

def register_hook(self, event_type: AppEventType, function: Callable) -> None:
    """
    A method that registers a function to run when an application event occurs.

    :param event_type: The application event type to listen for.
    :param function: The callback function to execute when the event occurs.
    """

    self.handler.register_hook(event_type, function)

register_task_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/app/manager.py

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

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

    self.task_handler.register_watcher(watcher)

unregister_config_callback

A method that unregisters a callback for configuration events.

Parameters:

• event_type (AppEventType) –

  The configuration event type.

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

  The callback function to remove.

Source code in dev_tool/event/app/manager.py

def unregister_config_callback(self, event_type: AppEventType, callback: Callable[[], None]) -> None:
    """
    A method that unregisters a callback for configuration events.

    :param event_type: The configuration event type.
    :param callback: The callback function to remove.
    """

    self.config_handler.unregister_callback(event_type, callback)

unregister_file_callback

A method that unregisters a callback for file events.

Parameters:

• path (Path) –

  The file path.

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

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

Source code in dev_tool/event/app/manager.py

def unregister_file_callback(self, path: Path, callback: Callable[[Path], None] | None = None) -> None:
    """
    A method that unregisters a callback for file events.

    :param path: The file path.
    :param callback: The callback function to remove, or None to remove all.
    """

    self.file_handler.unregister_callback(path, callback)

unregister_task_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/app/manager.py

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

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

    self.task_handler.unregister_watcher(watcher)