Skip to content

manager

dev_tool.services.notification.manager

NotificationManager

A manager class that coordinates between notification service and display.

This class orchestrates notification monitoring, display, and lifecycle without mixing the concerns of data management and presentation.

The constructor for the NotificationManager class.

Parameters:

Source code in dev_tool/services/notification/manager.py
def __init__(self, display: NotificationDisplay, service: NotificationService) -> None:
    """
    The constructor for the NotificationManager class.

    :param display: The notification display instance for rendering.
    :param service: The notification service instance for data management.
    """

    self.display = display
    self.last_count = 0
    self.last_update = 0
    self.service = service

display = display instance-attribute

last_count = 0 instance-attribute

last_update = 0 instance-attribute

service = service instance-attribute

clear

A method that clears all notifications.

Source code in dev_tool/services/notification/manager.py
def clear(self) -> None:
    """A method that clears all notifications."""

    self.service.clear()

display_active

A method that displays all active notifications in command mode.

Source code in dev_tool/services/notification/manager.py
def display_active(self) -> None:
    """A method that displays all active notifications in command mode."""

    notifications = self.service.get_active()

    if notifications:
        print(self.display.terminal.clear_eos, end='', flush=True)  # noqa: T201

        for notification in notifications:
            banner = self.display.format_notification_banner(notification)
            print(banner)  # noqa: T201

display_new

A method that displays any new notifications that haven't been shown yet.

Parameters:

  • displayed (set) –

    Set of already displayed notification keys.

Source code in dev_tool/services/notification/manager.py
def display_new(self, displayed: set) -> None:
    """
    A method that displays any new notifications that haven't been shown yet.

    :param displayed: Set of already displayed notification keys.
    """

    for notification in self.service.get_active():
        key = (notification.message, notification.timestamp)

        if key not in displayed:
            banner = self.display.format_notification_banner(notification)
            print(banner, flush=True)  # noqa: T201
            displayed.add(key)

error

A method that adds an error notification.

Parameters:

  • message (str) –

    The error message text.

Source code in dev_tool/services/notification/manager.py
def error(self, message: str) -> None:
    """
    A method that adds an error notification.

    :param message: The error message text.
    """

    self.service.error(message)

get_active_count

A method that gets the count of active notifications.

Returns:

  • int

    The number of active notifications.

Source code in dev_tool/services/notification/manager.py
def get_active_count(self) -> int:
    """
    A method that gets the count of active notifications.

    :return: The number of active notifications.
    """

    amount = self.service.get_active()
    return len(amount)

info

A method that adds an info notification.

Parameters:

  • message (str) –

    The info message text.

Source code in dev_tool/services/notification/manager.py
def info(self, message: str) -> None:
    """
    A method that adds an info notification.

    :param message: The info message text.
    """

    self.service.info(message)

is_changed

A method that checks if notifications have changed in a way that requires attention.

Returns:

  • bool

    True if notifications have changed, False otherwise.

Source code in dev_tool/services/notification/manager.py
def is_changed(self) -> bool:
    """
    A method that checks if notifications have changed in a way that requires attention.

    :return: True if notifications have changed, False otherwise.
    """

    current_second = int(time.time())
    current_count = len(self.service.get_active())

    is_changed = False

    if current_count != self.last_count:
        is_changed = True

    if current_count > 0 and current_second != self.last_update:
        is_changed = True

    if is_changed:
        self.last_update = current_second
        self.last_count = current_count

    return is_changed

monitor

A context manager for monitoring and displaying notifications during command execution.

Returns:

  • Generator[None, None, None]

    A generator for the context manager.

Source code in dev_tool/services/notification/manager.py
@contextmanager
def monitor(self) -> Generator[None, None, None]:
    """
    A context manager for monitoring and displaying notifications during command execution.

    :return: A generator for the context manager.
    """

    event = threading.Event()
    displayed = set()

    def _monitor() -> None:
        while not event.is_set():
            self.display_new(displayed)
            time.sleep(0.5)

    thread = threading.Thread(target=_monitor, daemon=True)
    thread.start()

    try:
        yield
    finally:
        event.set()
        thread.join(timeout=0.5)

normal

A method that adds a normal notification.

Parameters:

  • message (str) –

    The notification message text.

Source code in dev_tool/services/notification/manager.py
def normal(self, message: str) -> None:
    """
    A method that adds a normal notification.

    :param message: The notification message text.
    """

    self.service.normal(message)

success

A method that adds a success notification.

Parameters:

  • message (str) –

    The success message text.

Source code in dev_tool/services/notification/manager.py
def success(self, message: str) -> None:
    """
    A method that adds a success notification.

    :param message: The success message text.
    """

    self.service.success(message)

warning

A method that adds a warning notification.

Parameters:

  • message (str) –

    The warning message text.

Source code in dev_tool/services/notification/manager.py
def warning(self, message: str) -> None:
    """
    A method that adds a warning notification.

    :param message: The warning message text.
    """

    self.service.warning(message)