Pretalx data

intbot/core/admin.py

@@ -1,6 +1,6 @@
 import json
 
-from core.models import DiscordMessage, Webhook
+from core.models import DiscordMessage, PretalxData, Webhook
 from django.contrib import admin
 from django.utils.html import format_html
 
@@ -67,5 +67,29 @@ def content_short(self, obj):
         return f"{obj.content[:10]}...{obj.content[-10:]}"
 
 
+class PretalxDataAdmin(admin.ModelAdmin):
+    list_display = [
+        "uuid",
+        "endpoint",
+        "created_at",
+        "modified_at",
+    ]
+    list_filter = ["created_at"]
+    readonly_fields = fields = [
+        "uuid",
+        "endpoint",
+        "pretty_content",
+        "created_at",
+        "modified_at",
+        "processed_at",
+    ]
+
+    def pretty_content(self, obj):
+        return format_html("<pre>{}</pre>", json.dumps(obj.content, indent=4))
+
+    pretty_content.short_description = "Content"
+
+
 admin.site.register(Webhook, WebhookAdmin)
 admin.site.register(DiscordMessage, DiscordMessageAdmin)
+admin.site.register(PretalxData, PretalxDataAdmin)

intbot/core/integrations/pretalx.py

@@ -0,0 +1,67 @@
+from typing import Any
+
+import httpx
+from core.models import PretalxData
+from django.conf import settings
+
+PRETALX_EVENT = "ep2025"
+base_url = f"https://pretalx.com/api/events/{PRETALX_EVENT}/"
+
+RESOURCES = {
+    # Questions need to be passed to include answers in the same endpoint,
+    # saving us later time with joining the answers.
+    PretalxData.PretalxEndpoints.submissions: "submissions?questions=all",
+    PretalxData.PretalxEndpoints.speakers: "speakers?questions=all",
+}
+
+
+JsonType = dict[str, Any]
+
+
+def fetch_pretalx_data(resource) -> list[JsonType]:
+    headers = {
+        "Authorization": f"Token {settings.PRETALX_API_TOKEN}",
+        "Content-Type": "application/json",
+    }
+
+    endpoint = RESOURCES[resource]
+    url = base_url + f"{endpoint}"
+
+    # Pretalx paginates the output, so we will need to do multiple requests and
+    # then merge mutliple pages to one big dictionary
+    res0 = []
+    data = {"next": url}
+    n = 0
+    while url := data["next"]:
+        n += 1
+        response = httpx.get(url, headers=headers)
+
+        if response.status_code != 200:
+            raise Exception(f"Error {response.status_code}: {response.text}")
+
+        data = response.json()
+        res0 += data["results"]
+
+    return res0
+
+
+def download_latest_submissions() -> PretalxData:
+    data = fetch_pretalx_data(PretalxData.PretalxEndpoints.submissions)
+
+    pretalx_data = PretalxData.objects.create(
+        endpoint=PretalxData.PretalxEndpoints.submissions,
+        content=data,
+    )
+
+    return pretalx_data
+
+
+def download_latest_speakers() -> PretalxData:
+    data = fetch_pretalx_data(PretalxData.PretalxEndpoints.speakers)
+
+    pretalx_data = PretalxData.objects.create(
+        endpoint=PretalxData.PretalxEndpoints.speakers,
+        content=data,
+    )
+
+    return pretalx_data

intbot/core/models.py

@@ -81,3 +81,32 @@ def summary(self) -> str:
 
     def __str__(self):
         return f"{self.uuid} {self.author}: {self.content[:30]}"
+
+
+class PretalxData(models.Model):
+    """
+    Table to store raw data download from pretalx for later parsing.
+
+    We first download data from pretalx to this table, and then fire a separate
+    background task that pulls data from this table and stores in separate
+    "business" tables, like "Proposal" or "Speaker".
+    """
+
+    class PretalxEndpoints(models.TextChoices):
+        submissions = "submissions", "Submissions"
+        speakers = "speakers", "Speakers"
+        schedule = "schedule", "Schedule"
+
+    uuid = models.UUIDField(default=uuid.uuid4)
+    endpoint = models.CharField(
+        max_length=255,
+        choices=PretalxEndpoints.choices,
+    )
+    content = models.JSONField()
+
+    created_at = models.DateTimeField(auto_now_add=True)
+    modified_at = models.DateTimeField(auto_now=True)
+    processed_at = models.DateTimeField(blank=True, null=True)
+
+    def __str__(self):
+        return f"{self.uuid}"

intbot/intbot/settings.py

@@ -197,6 +197,9 @@ def get(name) -> str:
 ZAMMAD_GROUP_SPONSORS = get("ZAMMAD_GROUP_SPONSORS")
 ZAMMAD_GROUP_GRANTS = get("ZAMMAD_GROUP_GRANTS")
 
+# Pretalx
+PRETALX_API_TOKEN = get("PRETALX_API_TOKEN")
+
 
 if DJANGO_ENV == "dev":
     DEBUG = True
@@ -282,6 +285,8 @@ def get(name) -> str:
     ZAMMAD_GROUP_HELPDESK = "TestZammad Helpdesk"
     ZAMMAD_GROUP_BILLING = "TestZammad Billing"
 
+    PRETALX_API_TOKEN = "Test-Pretalx-API-token"
+
 
 elif DJANGO_ENV == "local_container":
     DEBUG = False

intbot/tests/test_integrations/test_pretalx.py

@@ -0,0 +1,165 @@
+import respx
+import pytest
+from core.integrations import pretalx
+from core.models import PretalxData
+from httpx import Response
+
+
+@respx.mock
+def test_fetch_submissions_from_pretalx():
+    endpoint = pretalx.RESOURCES[PretalxData.PretalxEndpoints.submissions]
+    url = pretalx.base_url + endpoint
+    respx.get(url).mock(
+        return_value=Response(
+            200,
+            json={
+                "results": [
+                    {"hello": "world"},
+                ],
+                "next": f"{url}&page=2",
+            },
+        )
+    )
+    respx.get(url + "&page=2").mock(
+        return_value=Response(
+            200,
+            json={
+                "results": [
+                    {"foo": "bar"},
+                ],
+                # It's important to make it last page in tests.
+                # Otherwise it will be infinite loop :)
+                "next": None,
+            },
+        )
+    )
+
+    submissions = pretalx.fetch_pretalx_data(
+        PretalxData.PretalxEndpoints.submissions,
+    )
+
+    assert submissions == [
+        {"hello": "world"},
+        {"foo": "bar"},
+    ]
+
+
+@respx.mock
+def test_fetch_speakers_from_pretalx():
+    endpoint = pretalx.RESOURCES[PretalxData.PretalxEndpoints.speakers]
+    url = pretalx.base_url + endpoint
+    respx.get(url).mock(
+        return_value=Response(
+            200,
+            json={
+                "results": [
+                    {"hello": "world"},
+                ],
+                "next": f"{url}&page=2",
+            },
+        )
+    )
+    respx.get(url + "&page=2").mock(
+        return_value=Response(
+            200,
+            json={
+                "results": [
+                    {"foo": "bar"},
+                ],
+                # It's important to make it last page in tests.
+                # Otherwise it will be infinite loop :)
+                "next": None,
+            },
+        )
+    )
+
+    submissions = pretalx.fetch_pretalx_data(
+        PretalxData.PretalxEndpoints.speakers,
+    )
+
+    assert submissions == [
+        {"hello": "world"},
+        {"foo": "bar"},
+    ]
+
+
+@respx.mock
+@pytest.mark.django_db
+def test_download_latest_submissions():
+    endpoint = pretalx.RESOURCES[PretalxData.PretalxEndpoints.submissions]
+    url = pretalx.base_url + endpoint
+    respx.get(url).mock(
+        return_value=Response(
+            200,
+            json={
+                "results": [
+                    {"hello": "world"},
+                ],
+                "next": f"{url}&page=2",
+            },
+        )
+    )
+    respx.get(url + "&page=2").mock(
+        return_value=Response(
+            200,
+            json={
+                "results": [
+                    {"foo": "bar"},
+                ],
+                # It's important to make it last page in tests.
+                # Otherwise it will be infinite loop :)
+                "next": None,
+            },
+        )
+    )
+
+    pretalx.download_latest_submissions()
+
+    pd = PretalxData.objects.get(endpoint=PretalxData.PretalxEndpoints.submissions)
+
+    assert pd.endpoint == "submissions"
+    assert pd.content == [
+        {"hello": "world"},
+        {"foo": "bar"},
+    ]
+
+@respx.mock
+@pytest.mark.django_db
+def test_download_latest_speakers():
+    endpoint = pretalx.RESOURCES[PretalxData.PretalxEndpoints.speakers]
+    url = pretalx.base_url + endpoint
+    respx.get(url).mock(
+        return_value=Response(
+            200,
+            json={
+                "results": [
+                    {"hello": "world"},
+                ],
+                "next": f"{url}&page=2",
+            },
+        )
+    )
+    respx.get(url + "&page=2").mock(
+        return_value=Response(
+            200,
+            json={
+                "results": [
+                    {"foo": "bar"},
+                ],
+                # It's important to make it last page in tests.
+                # Otherwise it will be infinite loop :)
+                "next": None,
+            },
+        )
+    )
+
+    pretalx.download_latest_speakers()
+
+    pd = PretalxData.objects.get(endpoint=PretalxData.PretalxEndpoints.speakers)
+
+    assert pd.endpoint == "speakers"
+    assert pd.content == [
+        {"hello": "world"},
+        {"foo": "bar"},
+    ]
+