Adding notifier to allow sending some kind of event to outside recipients (as emails, telegram, ...)

server/src/uds/core/alerts/notifier.py

@@ -29,6 +29,7 @@
 .. moduleauthor:: Adolfo Gómez, dkmaster at dkmon dot com
 """
 import typing
+import enum
 
 from django.utils.translation import gettext_noop as _
 
@@ -37,6 +38,16 @@ from uds.core import Module
 if typing.TYPE_CHECKING:
     from uds.core.environment import Environment
 
+class NotifierLevel(enum.IntEnum):
+    """
+    Notifier levels
+    """
+    INFO = 0
+    WARNING = 1
+    ERROR = 2
+    CRITICAL = 3
+
+
 class Notifier(Module):
     """
     this class provides an abstraction of a notifier system for administrator defined events
@@ -67,6 +78,7 @@ class Notifier(Module):
     # : your own :py:meth:uds.core.module.BaseModule.icon method.
     iconFile: typing.ClassVar[str] = 'notifier.png'
 
+    level: NotifierLevel = NotifierLevel.ERROR
 
     def __init__(self, environment: 'Environment', values: Module.ValuesType):
         super().__init__(environment, values)
@@ -88,6 +100,17 @@ class Notifier(Module):
         Default implementation does nothing
         """
 
-    def notify(self, title: str, message: str) -> bool:
-        return False
+    def notify(self, level: NotifierLevel,  message: str, *args, **kwargs) -> None:
+        """
+        This method will be invoked from UDS to notify an event to this notifier.
+        This method will be invoked in real time, so ensure this method does not block or
+        do any long operations. (use threading if you need to do that)
+
+        :param level: Level of event
+        :param message: Message to be shown
+        :param args: Arguments to be used in message
+        :param kwargs: Keyword arguments to be used in message
+        :return: None
+        """ 
+        pass
 

server/src/uds/migrations/0044_remove_authenticator_visible_and_more.py

@@ -1,4 +1,4 @@
-# Generated by Django 4.0 on 2022-02-08 19:08
+# Generated by Django 4.0 on 2022-02-08 19:25
 
 from django.db import migrations, models
 import django.db.models.deletion
@@ -82,7 +82,7 @@ class Migration(migrations.Migration):
                 ('name', models.CharField(default='', max_length=128)),
                 ('comments', models.CharField(default='', max_length=256)),
                 ('enabled', models.BooleanField(default=True)),
-                ('level', models.CharField(choices=[('INFO', 'Info'), ('WARNING', 'Warning'), ('ERROR', 'Error'), ('CRITICAL', 'Critical')], default='ERROR', max_length=16)),
+                ('level', models.PositiveSmallIntegerField(choices=[(1, 'Info'), (2, 'Warning'), (3, 'Error'), (4, 'Critical')], default=1)),
                 ('tags', models.ManyToManyField(to='uds.Tag')),
             ],
             options={

server/src/uds/models/notifier.py

@@ -34,36 +34,23 @@ import typing
 
 from django.db import models
 
+from uds.core.alerts import notifier
+from uds.core.util.singleton import Singleton
+from uds.core.workers import initialize
+
 from .managed_object_model import ManagedObjectModel
 from .tag import TaggingMixin
 
 logger = logging.getLogger(__name__)
 
-if typing.TYPE_CHECKING:
-    from uds.core.alerts import notifier
-
 
 class Notifier(ManagedObjectModel, TaggingMixin):
 
-    NOTIFIER_LEVEL_INFO = 'INFO'
-    NOTIFIER_LEVEL_WARNING = 'WARNING'
-    NOTIFIER_LEVEL_ERROR = 'ERROR'
-    NOTIFIER_LEVEL_CRITICAL = 'CRITICAL'
-
-    NOTIFIER_LEVEL_CHOICES = (
-        (NOTIFIER_LEVEL_INFO, 'Info'),
-        (NOTIFIER_LEVEL_WARNING, 'Warning'),
-        (NOTIFIER_LEVEL_ERROR, 'Error'),
-        (NOTIFIER_LEVEL_CRITICAL, 'Critical'),
-    )
-
     name = models.CharField(max_length=128, default='')
     comments = models.CharField(max_length=256, default='')
     enabled = models.BooleanField(default=True)
-    level = models.CharField(
-        max_length=16,
-        choices=NOTIFIER_LEVEL_CHOICES,
-        default=NOTIFIER_LEVEL_ERROR
+    level = models.PositiveSmallIntegerField(
+        default=notifier.NotifierLevel.ERROR,
     )
 
     class Meta:
@@ -78,3 +65,39 @@ class Notifier(ManagedObjectModel, TaggingMixin):
         self, values: typing.Optional[typing.Dict[str, str]] = None
     ) -> 'notifier.Notifier':
         return typing.cast('notifier.Notifier', super().getInstance(values=values))
+
+
+class Notifiers(Singleton):
+    """
+    This class is a singleton that contains all notifiers, so we can
+    easily notify to all of them.
+    """
+    notifiers: typing.Dict[str, 'notifier.Notifier'] = {}
+    initialized: bool = False
+
+    def __init__(self):
+        super().__init__()
+        self.notifiers: typing.Dict[str, 'notifier.Notifier'] = {}
+    
+    def reload(self) -> None:
+        """
+        Loads all notifiers from db.
+        """
+        for n in Notifier.objects.filter(enabled=True):
+            self.notifiers[n.name] = n.getInstance()
+
+    def notify(self, level: 'notifier.NotifierLevel', message: str, *args, **kwargs) -> None:
+        """
+        Notifies all notifiers with level equal or higher than given level.
+
+        :param level: Level to notify
+        :param message: Message to notify
+        :return: None
+        """
+        # initialize notifiers if needed
+        if not self.initialized:
+            self.reload()
+            self.initialized = True
+
+        for n in (n for n in self.notifiers.values() if n.level >= level):
+            n.notify(level, message, *args, **kwargs)