PyCS/pycs/frontend/notifications/NotificationManager.py

import warnings

from functools import partial
from socketio import Server
from typing import Callable

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
from pycs.util import Singleton


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

    def __repr__(self) -> str:
        return f"<{self.__class__.__name__} 0x{id(self):x}>"

    def setup(self, sio: Server):
        self.sio = sio

    def init(self):
        self.sio = None
        self.json = JSONEncoder()

    def __call__(self, name, obj_id, cls=None):
        if self.sio is None:
            warnings.warn("NotificationManager was called before a setup was performed!")
            return

        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 _new_signal(self, signal_name, suffix, *args, **kwargs):
        return partial(self, f'{signal_name}-{suffix}', *args, **kwargs)


    def create(self, suffix, *args, **kwargs) -> Callable:
        return self._new_signal('create', suffix, *args, **kwargs)


    def edit(self, suffix, *args, **kwargs) -> Callable:
        return self._new_signal('edit', suffix, *args, **kwargs)


    def remove(self, suffix, *args, **kwargs) -> Callable:
        return self._new_signal('remove', suffix, *args, **kwargs)


    @classmethod
    def removed(kls, suffix, *args, **kwargs):
        return kls().remove(suffix)(*args, **kwargs)


    @classmethod
    def edited(kls, suffix, *args, **kwargs):
        return kls().edit(suffix)(*args, **kwargs)


    @classmethod
    def created(kls, suffix, *args, **kwargs):
        return kls().create(suffix)(*args, **kwargs)