Source code for divera.triggers.socketevents

import abc
import logging
import traceback
import typing

import divera.utils
from divera.models.alarms import Alarm
from divera.models.events import Event
from divera.models.message import Message
from divera.models.message_channel import MessageChannel
from divera.models.news import News

from . import Trigger


class SocketEvent(Trigger):
    def __init__(
            self,
            data,
    ):
        self.data = data

    @staticmethod
    @abc.abstractmethod
    def matches(data) -> bool:
        return False


class UnspecifiedSocketEvent(SocketEvent):
    @staticmethod
    def matches(data) -> bool:
        return True


class ClusterMonitorEvent(SocketEvent):
    @staticmethod
    def matches(data) -> bool:
        return data['type'] == 'cluster-monitor'


class ClusterPullTrigger(SocketEvent):
    models = None
    action_path = [
        'pull',
        'action',
    ]

    @property
    def action(self):
        d = self.data
        try:
            for step in self.action_path:
                d = d[step]
        except KeyError:
            return None
        return d

    @staticmethod
    def matches(data) -> bool:
        return data['type'] == 'cluster-pull'


class ClusterPullAlarm(ClusterPullTrigger):
    models = (
        Alarm,
    )

    @staticmethod
    def matches(data) -> bool:
        return ClusterPullTrigger.matches(data=data) and data['pull']['type'] == 'alarm'


class ClusterPullEvent(ClusterPullTrigger):
    models = (
        Event,
    )

    @staticmethod
    def matches(data) -> bool:
        return ClusterPullTrigger.matches(data=data) and data['pull']['type'] == 'event'


class ClusterTrigger(ClusterPullTrigger, abc.ABC):
    @property
    @abc.abstractmethod
    def data_path(self) -> list:
        return [
        ]

    @property
    def object(self):
        if type(self.models) in [list, tuple]:
            model = self.models[0]
        else:
            model = self.models
        data = self.data
        for p in self.data_path:
            data = data[p]
        return model(
            data=data,
        )


class ClusterMessage(ClusterTrigger):
    models = (
        Message,
    )
    data_path = [
        'message',
    ]

    @staticmethod
    def matches(data) -> bool:
        """
        Matches :code:`user-message` and :code:`cluster-message`.

        * :code:`user-message` is triggered when a message is received in a message channel that is related to something
        * :code:`cluster-message` is triggered when a message is received in the default message channel

        """
        return data['type'] in ['user-message', 'cluster-message'] and 'message' in data.keys()

    @property
    def message(self) -> Message:
        if 'message' in self.data:
            return Message(
                data=self.data['message'],
            )

    @property
    def message_channel(self) -> MessageChannel:
        if 'message_channel' in self.data:
            return MessageChannel(
                data=self.data['message_channel'],
            )


class MessageDeletedEvent(ClusterMessage):
    @staticmethod
    def matches(data) -> bool:
        return ClusterMessage.matches(data) and data['message']['deleted']


class ClusterPullNews(ClusterPullTrigger):
    models = (
        News,
    )

    @staticmethod
    def matches(data) -> bool:
        return ClusterPullTrigger.matches(data=data) and data['pull']['type'] == 'news'


class ClusterPullMessageChannel(ClusterPullTrigger):
    models = (
        MessageChannel,
    )

    @staticmethod
    def matches(data) -> bool:
        return ClusterPullTrigger.matches(data=data) and data['pull']['type'] == 'message-channel'


class UnspecifiedClusterPull(ClusterPullTrigger):
    models = None


class UserStatusChange(SocketEvent):
    @staticmethod
    def matches(data) -> bool:
        return data['type'] == 'user-status'


def get_matching_subclasses(
        data,
        cls: SocketEvent = SocketEvent,
        sort: bool = True,
) -> typing.List[typing.Type[SocketEvent]]:
    subclasses = []
    for s_cls in cls.__subclasses__():
        for r in get_matching_subclasses(
                data=data,
                cls=s_cls,
                sort=False,
        ):
            subclasses.append(r)
    try:
        if abc.ABC not in cls.__bases__:
            if cls.matches(data):
                subclasses.append(cls)
    except Exception as e:
        logging.error(
            f'{cls.__name__} is no match for data of type {type(data).__name__}',
        )
        logging.error(''.join(traceback.format_exception(e)))
    if sort:
        return sorted(  # Sort most specific (= most deeply nested) classes first
            sorted(  # Sort unspecified classes last
                subclasses,
                key=lambda cls_: cls_.__name__.startswith('Unspecified'),
            ),
            key=lambda x: divera.utils.get_nesting_depth(x, parent_cls=SocketEvent),
            reverse=True,
        )
    else:
        return subclasses