PyCS/pycs/frontend/notifications/NotificationManager.py

from socketio import Server

from pycs import app
from pycs.database.File import File
from pycs.database.Label import Label
from pycs.database.Model import Model
from pycs.database.Project import Project
from pycs.database.Result import Result
from pycs.frontend.util.JSONEncoder import JSONEncoder
from pycs.jobs.Job import Job


class NotificationManager:
    """
    send events via a socket.io connection
    """

    def __init__(self, sio: Server):
        self.sio = sio
        self.json = JSONEncoder()

    def __emit(self, name, obj_id, cls=None):
        if cls is not None:
            assert isinstance(obj_id, int), \
                f"{cls.__name__} ID must be an integer, but was {type(obj_id)} ({obj_id=})!"
            obj = cls.query.get(obj_id)

        else:
            obj = obj_id

        assert obj is not None, "Object was unexpectedly None!"
        app.logger.debug(f"{name}: {obj}")

        enc = self.json.default(obj)
        self.sio.emit(name, enc)

    def create_job(self, created_job: Job):
        """
        fire create-job event

        :param created_job:
        :return:
        """
        self.__emit('create-job', created_job)

    def edit_job(self, edited_job: Job):
        """
        fire edit-job event

        :param edited_job:
        :return:
        """
        self.__emit('edit-job', edited_job)

    def remove_job(self, removed_job: Job):
        """
        fire remove-job event

        :param removed_job:
        :return:
        """
        self.__emit('remove-job', removed_job)

    def create_model(self, created_model_id: int):
        """
        fire create-model event

        :param created_model:
        :return:
        """
        self.__emit('create-model', created_model_id, Model)

    def remove_model(self, model_serialized: dict):
        """
        fire remove-model event

        :param removed_model:
        :return:
        """
        self.__emit('remove-model', model_serialized)

    def create_project(self, created_project_id: int):
        """
        fire create-project event

        :param created_project:
        :return:
        """
        self.__emit('create-project', created_project_id, Project)

    def remove_project(self, project_serialized: dict):
        """
        fire remove-project event

        :param removed_project:
        :return:
        """
        self.__emit('remove-project', project_serialized)

    def edit_project(self, edited_project_id: int):
        """
        fire edit-project event

        :param edited_project:
        :return:
        """
        self.__emit('edit-project', edited_project_id, Project)

    def create_label(self, created_label_id: int):
        """
        fire create-label event

        :param created_label:
        :return:
        """
        self.__emit('create-label', created_label_id, Label)

    def edit_label(self, edited_label_id: int):
        """
        fire edit-label event

        :param edited_label:
        :return:
        """
        self.__emit('edit-label', edited_label_id, Label)

    def remove_label(self, label_serialized: dict):
        """
        fire remove-label event

        :param removed_label:
        :return:
        """
        self.__emit('remove-label', label_serialized)

    def create_file(self, created_file_id: int):
        """
        fire create-file event

        :param created_file:
        :return:
        """
        self.__emit('create-file', created_file_id, File)

    def edit_file(self, edited_file_id: int):
        """
        fire edit-file event

        :param edited_file:
        :return:
        """
        self.__emit('edit-file', edited_file_id, File)

    def remove_file(self, file_serialized: dict):
        """
        fire remove-file event

        :param removed_file:
        :return:
        """
        self.__emit('remove-file', file_serialized)

    def create_result(self, created_result_id: int):
        """
        fire create-result event

        :param created_result:
        :return:
        """
        self.__emit('create-result', created_result_id, Result)

    def edit_result(self, edited_result_id: int):
        """
        fire edit-result event

        :param edited_result:
        :return:
        """
        self.__emit('edit-result', edited_result_id, Result)

    def remove_result(self, result_serialized: dict):
        """
        fire remove-result event

        :param removed_result:
        :return:
        """
        self.__emit('remove-result', result_serialized)