feat: add support for multiple notification setups

Author: janezicmatej

configuration/config.py

@@ -13,6 +13,7 @@
     NotificationGeneric,
     NotificationSlack,
     NotificationTelegram,
+    TelegramBot,
 )
 
 
@@ -67,35 +68,34 @@ def get_epoch(chain_id: int) -> Epoch:
 
 
 def get_notification_config() -> Notification:
-    discord = None
     discord_webhook = os.environ.get("NOTIFICATION_DISCORD_WEBHOOK")
+    discord = []
     if discord_webhook is not None:
-        discord = NotificationDiscord(discord_webhook)
+        discord.extend(discord_webhook.split(","))
 
-    slack = None
     slack_webhook = os.environ.get("NOTIFICATION_SLACK_WEBHOOK")
+    slack = []
     if slack_webhook is not None:
-        slack = NotificationSlack(slack_webhook)
+        slack.extend(slack_webhook.split(","))
 
-    telegram = None
     telegram_bot_token = os.environ.get("NOTIFICATION_TELEGRAM_BOT_TOKEN")
     telegram_chat_id = os.environ.get("NOTIFICATION_TELEGRAM_CHAT_ID")
+    telegram = []
     if telegram_bot_token is not None and telegram_chat_id is not None:
-        telegram = NotificationTelegram(
-            bot_token=telegram_bot_token,
-            chat_id=telegram_chat_id,
-        )
+        bot_tokens = telegram_bot_token.split(",")
+        chat_ids = telegram_chat_id.split(",")
+        telegram = [TelegramBot(t, c) for t, c in zip(bot_tokens, chat_ids)]
 
-    generic = None
     generic_webhook = os.environ.get("NOTIFICATION_GENERIC_WEBHOOK")
+    generic = []
     if generic_webhook is not None:
-        generic = NotificationGeneric(generic_webhook)
+        generic.extend(generic_webhook.split(","))
 
     return Notification(
-        discord=discord,
-        slack=slack,
-        telegram=telegram,
-        generic=generic,
+        discord=NotificationDiscord(discord),
+        slack=NotificationSlack(slack),
+        telegram=NotificationTelegram(telegram),
+        generic=NotificationGeneric(generic),
     )
 
 

configuration/types.py

@@ -184,31 +184,36 @@ class Epoch:
 
 @frozen
 class NotificationDiscord:
-    webhook_url: str
+    webhook_url: list[str]
 
 
 @frozen
 class NotificationSlack:
-    webhook_url: str
+    webhook_url: list[str]
 
 
 @frozen
-class NotificationTelegram:
+class TelegramBot:
     bot_token: str
     chat_id: str
 
 
+@frozen
+class NotificationTelegram:
+    bot: list[TelegramBot]
+
+
 @frozen
 class NotificationGeneric:
-    webhook_url: str
+    webhook_url: list[str]
 
 
 @frozen
 class Notification:
-    discord: NotificationDiscord | None
-    slack: NotificationSlack | None
-    telegram: NotificationTelegram | None
-    generic: NotificationGeneric | None
+    discord: NotificationDiscord
+    slack: NotificationSlack
+    telegram: NotificationTelegram
+    generic: NotificationGeneric
 
 
 @frozen

observer/notification.py

@@ -26,43 +26,41 @@ def notify(
         pass
 
 
-def notify_discord(
-    config: NotificationDiscord, message: str
-) -> requests.Response | None:
-    return notify(
-        config.webhook_url,
-        "POST",
-        headers={"Content-Type": "application/json"},
-        json={"content": message},
-    )
+def notify_discord(config: NotificationDiscord, message: str) -> None:
+    for u in config.webhook_url:
+        notify(
+            u,
+            "POST",
+            headers={"Content-Type": "application/json"},
+            json={"content": message},
+        )
 
 
-def notify_slack(config: NotificationSlack, message: str) -> requests.Response | None:
-    return notify(
-        config.webhook_url,
-        "POST",
-        headers={"Content-Type": "application/json"},
-        json={"text": message},
-    )
+def notify_slack(config: NotificationSlack, message: str) -> None:
+    for u in config.webhook_url:
+        notify(
+            u,
+            "POST",
+            headers={"Content-Type": "application/json"},
+            json={"text": message},
+        )
 
 
-def notify_telegram(
-    config: NotificationTelegram, message: str
-) -> requests.Response | None:
-    return notify(
-        f"https://api.telegram.org/bot{config.bot_token}/sendMessage",
-        "POST",
-        headers={"Content-Type": "application/json"},
-        json={"chat_id": config.chat_id, "text": message},
-    )
+def notify_telegram(config: NotificationTelegram, message: str) -> None:
+    for t in config.bot:
+        notify(
+            f"https://api.telegram.org/bot{t.bot_token}/sendMessage",
+            "POST",
+            headers={"Content-Type": "application/json"},
+            json={"chat_id": t.chat_id, "text": message},
+        )
 
 
-def notify_generic(
-    config: NotificationGeneric, issue: "Message"
-) -> requests.Response | None:
-    return notify(
-        config.webhook_url,
-        "POST",
-        headers={"Content-Type": "application/json"},
-        json={"level": issue.level.value, "message": issue.message},
-    )
+def notify_generic(config: NotificationGeneric, issue: "Message") -> None:
+    for u in config.webhook_url:
+        notify(
+            u,
+            "POST",
+            headers={"Content-Type": "application/json"},
+            json={"level": issue.level.value, "message": issue.message},
+        )

observer/observer.py

@@ -182,17 +182,12 @@ def log_message(config: Configuration, message: Message):
 
     n = config.notification
 
-    if n.discord is not None:
-        notify_discord(n.discord, message.level.name + " " + message.message)
+    lvl_msg = f"{message.level.name} {message.message}"
 
-    if n.slack is not None:
-        notify_slack(n.slack, message.level.name + " " + message.message)
-
-    if n.telegram is not None:
-        notify_telegram(n.telegram, message.level.name + " " + message.message)
-
-    if n.generic is not None:
-        notify_generic(n.generic, message)
+    notify_discord(n.discord, lvl_msg)
+    notify_slack(n.slack, lvl_msg)
+    notify_telegram(n.telegram, lvl_msg)
+    notify_generic(n.generic, message)
 
 
 async def observer_loop(config: Configuration) -> None: