baseimage-tutorial-framework/tfw/components/frontend/message_storage.py

from abc import ABC, abstractmethod
from contextlib import suppress

from .message_sender import MessageSender


class MessageStorage(ABC):
    def __init__(self):
        self._messages = []

    def save_message(self, message):
        with suppress(KeyError, AttributeError):
            if self._filter_message(message):
                self._messages.extend(self._transform_message(message))

    @abstractmethod
    def _filter_message(self, message):
        raise NotImplementedError

    def _transform_message(self, message):   # pylint: disable=no-self-use
        yield message

    def clear(self):
        self._messages.clear()

    @property
    def messages(self):
        yield from self._messages


class FrontendMessageStorage(MessageStorage):
    def _filter_message(self, message):
        return message['key'].startswith((
            'console.write',
            'dashboard.layout',
            'dashboard.terminalMenuItem',
            'message.send',
            'message.config',
            'ide.read'
        ))

    def _transform_message(self, message):
        transformations = {
            'message.queue': MessageSender.generate_messages_from_queue,
            'ide.read': self._delete_ide_content
        }
        if message['key'] in transformations:
            yield from transformations[message['key']](message)

    @staticmethod
    def _delete_ide_content(message):
        del message['content']
        yield message