initial router placeholder

Author: artcz

intbot/core/bot/channel_router.py

@@ -0,0 +1,83 @@
+"""
+Channel router that decides where to send particular message
+"""
+
+from dataclasses import dataclass
+from core.models import Webhook
+
+from core.integrations.github import GithubProjects, GithubRepositories
+from core.integrations.github import GithubWebhook
+from django.conf import settings
+
+@dataclass
+class DiscordChannel:
+    channel_id: int
+    channel_name: str
+
+
+
+dont_send_it = DiscordChannel(channel_id=0, channel_name="/dev/null")
+
+class Channels:
+    test_channel = DiscordChannel(
+        channel_id=settings.DISCORD_TEST_CHANNEL_ID,
+        channel_name=settings.DISCORD_TEST_CHANNEL_NAME,
+    )
+
+    board_channel = DiscordChannel(channel_id=..., channel_name=...)
+    ep2025_channel = DiscordChannel(channel_id=..., channel_name=...)
+    em_channel = DiscordChannel(channel_id=..., channel_name=...)
+    website_channel = DiscordChannel(channel_id=..., channel_name=...)
+    bot_channel = DiscordChannel(channel_id=..., channel_name=...)
+
+
+def discord_channel_router(wh: Webhook) -> DiscordChannel:
+
+    if wh.source == "github":
+        return github_router(wh)
+
+    elif wh.source == "internal":
+        return internal_router(wh)
+
+    return dont_send_it
+
+
+def github_router(wh: Webhook) -> DiscordChannel:
+
+    gwh = GithubWebhook.from_webhook(wh)
+    project = gwh.get_project()
+    repository = gwh.get_repository()
+
+    # We have three github projects, that we want to route to three different
+    # channels - one for ep2025, one for EM, and one for the board.
+    if project == GithubProjects.board_project:
+        return Channels.board_channel
+
+    elif project == GithubProjects.ep2025_project:
+        return Channels.ep2025_channel
+
+    elif project == GithubProjects.em_project:
+        return Channels.em_channel
+
+    else:
+        ...
+
+    # Then we have our open source repositories, like website, this bot, and
+    # some others, that we also might want to route to different channels
+    if repository == GithubRepositories.website_repo:
+        return Channels.website_channel
+
+    elif repository == GithubRepositories.bot_repo:
+        return Channels.bot_channel
+
+    elif repository == ...:
+        ...
+
+    # Finally, we can use this to drop notifications that we don't want to
+    # support, by routing them to /dev/null
+    return dont_send_it
+
+
+def internal_router(wh: Webhook) -> DiscordChannel:
+    # For now just send all the internal messages to a test channel
+    return Channels.test_channel

intbot/core/integrations/github.py

@@ -3,8 +3,39 @@
 import httpx
 from django.conf import settings
 
+from core.models import Webhook
+
 GITHUB_API_URL = "https://api.github.com/graphql"
 
+
+class GithubRepositories:
+    website_repo = ...
+    bot_repo = ...
+
+
+class GithubProjects:
+    board_project = ...
+    ep2025_project = ...
+    em_project = ...
+
+
+class GithubWebhook:
+
+    def __init__(self, headers, content):
+        self.headers = headers
+        self.content = content
+
+    @classmethod
+    def from_webhook(cls, wh: Webhook):
+        return cls(headers=wh.meta, content=wh.content)
+
+    def get_project(self):
+        ...
+
+    def get_repository(self):
+        ...
+
+
 # GraphQL query
 query = """
 query($itemId: ID!) {

intbot/core/tasks.py

@@ -1,8 +1,8 @@
 import logging
 
 from core.integrations.github import parse_github_webhook
+from core.bot.channel_router import discord_channel_router
 from core.models import DiscordMessage, Webhook
-from django.conf import settings
 from django.utils import timezone
 from django_tasks import task
 
@@ -27,9 +27,13 @@ def process_internal_webhook(wh: Webhook):
     if wh.source != "internal":
         raise ValueError("Incorrect wh.source = {wh.source}")
 
+    channel = discord_channel_router(wh)
+
     DiscordMessage.objects.create(
-        channel_id=settings.DISCORD_TEST_CHANNEL_ID,
-        channel_name=settings.DISCORD_TEST_CHANNEL_NAME,
+        channel_id=channel.channel_id,
+        channel_name=channel.channel_name,
+        # channel_id=settings.DISCORD_TEST_CHANNEL_ID,
+        # channel_name=settings.DISCORD_TEST_CHANNEL_NAME,
         content=f"Webhook content: {wh.content}",
         # Mark as not sent - to be sent with the next batch
         sent_at=None,
@@ -51,10 +55,13 @@ def process_github_webhook(wh: Webhook):
         logger.info(f"Not processing Github Webhook {wh.uuid}: {e}")
         return
 
-    # NOTE WHERE SHOULD WE GET THE CHANNEL ID FROM?
+    channel = discord_channel_router(wh)
+
     DiscordMessage.objects.create(
-        channel_id=settings.DISCORD_TEST_CHANNEL_ID,
-        channel_name=settings.DISCORD_TEST_CHANNEL_NAME,
+        channel_id=channel.channel_id,
+        channel_name=channel.channel_name,
+        # channel_id=settings.DISCORD_TEST_CHANNEL_ID,
+        # channel_name=settings.DISCORD_TEST_CHANNEL_NAME,
         content=f"GitHub: {message}",
         # Mark as unsend - to be sent with the next batch
         sent_at=None,

intbot/intbot/settings.py

@@ -116,6 +116,39 @@
 # Just to make mypy happy
 TASKS: dict[str, Any]
 
+
+# There are bunch of settings that we can skip on dev/testing environments if
+# are not used, but should be present on prod/staging deployments.
+# Instead of repeating them per-env below, they go here.
+def get(name) -> str:
+    value = os.environ.get(name)
+
+    if DJANGO_ENV == "prod":
+        if value is None:
+            raise ValueError(f"{name} is not set")
+
+    elif DJANGO_ENV == "test":
+        pass
+    else:
+        warnings.warn(f"{name} not set")
+
+    return value or ""
+
+# Discord
+# This is only needed if you end up running the bot locally, hence it
+# doesn't fail explicilty – however it does emit a warning.
+# Please check the documentation and/or current online guides how to get
+# one from the developer portal.
+# If you run it locally, you probably want to run it against your own test
+# bot and a test server.
+DISCORD_TEST_CHANNEL_ID = get("DISCORD_TEST_CHANNEL_ID")
+DISCORD_TEST_CHANNEL_NAME = get("DISCORD_TEST_CHANNEL_NAME")
+DISCORD_BOT_TOKEN = get("DISCORD_BOT_TOKEN")
+
+# Github
+GITHUB_API_TOKEN = get("GITHUB_API_TOKEN")
+GITHUB_WEBHOOK_SECRET_TOKEN = get("GITHUB_WEBHOOK_SECRET_TOKEN")
+
 if DJANGO_ENV == "dev":
     DEBUG = True
     ALLOWED_HOSTS = ["127.0.0.1", "localhost"]
@@ -141,23 +174,6 @@
 
     WEBHOOK_INTERNAL_TOKEN = "dev-token"
 
-    # This is only needed if you end up running the bot locally, hence it
-    # doesn't fail explicilty – however it does emit a warning.
-    # Please check the documentation and/or current online guides how to get
-    # one from the developer portal.
-    # If you run it locally, you probably want to run it against your own test
-    # bot and a test server.
-
-    def warn_if_missing(name, default=""):
-        value = os.environ.get(name, default)
-        if not value:
-            warnings.warn(f"{name} not set")
-
-    DISCORD_TEST_CHANNEL_ID = warn_if_missing("DISCORD_TEST_CHANNEL_ID", "")
-    DISCORD_TEST_CHANNEL_NAME = warn_if_missing("DISCORD_TEST_CHANNEL_NAME", "")
-    DISCORD_BOT_TOKEN = warn_if_missing("DISCORD_BOT_TOKEN", "")
-    GITHUB_API_TOKEN = warn_if_missing("GITHUB_API_TOKEN", "")
-    GITHUB_WEBHOOK_SECRET_TOKEN = warn_if_missing("GITHUB_WEBHOOK_SECRET_TOKEN", "")
 
 
 elif DJANGO_ENV == "test":
@@ -293,17 +309,10 @@ def warn_if_missing(name, default=""):
         },
     }
 
-    DISCORD_BOT_TOKEN = os.environ["DISCORD_BOT_TOKEN"]
-    DISCORD_TEST_CHANNEL_ID = os.environ["DISCORD_TEST_CHANNEL_ID"]
-    DISCORD_TEST_CHANNEL_NAME = os.environ["DISCORD_TEST_CHANNEL_NAME"]
-
     CSRF_TRUSTED_ORIGINS = [
         "https://internal.europython.eu",
     ]
 
-    WEBHOOK_INTERNAL_TOKEN = os.environ["WEBHOOK_INTERNAL_TOKEN"]
-    GITHUB_API_TOKEN = os.environ["GITHUB_API_TOKEN"]
-    GITHUB_WEBHOOK_SECRET_TOKEN = os.environ["GITHUB_WEBHOOK_SECRET_TOKEN"]
 
 
 elif DJANGO_ENV == "build":