download pretix data

Author: artcz

intbot/core/integrations/pretix.py

@@ -0,0 +1,81 @@
+import logging
+from typing import Any
+
+import httpx
+from core.models import PretixData
+from django.conf import settings
+
+logger = logging.getLogger(__name__)
+
+PRETALX_EVENTS = [
+    "2022",
+    "ep2023",
+    "ep2024",
+    "ep2025",
+]
+
+ENDPOINTS = {
+    # Questions need to be passed to include answers in the same endpoint,
+    # saving us later time with joining the answers.
+    PretixData.PretixResources.orders: "orders/",
+    PretixData.PretixResources.products: "items/",
+    PretixData.PretixResources.vouchers: "vouchers/",
+}
+
+
+JsonType = dict[str, Any]
+
+
+def get_event_url(event):
+    assert event in PRETALX_EVENTS
+
+    pretix_url = f"https://tickets.europython.eu"
+    url = f"{pretix_url}/api/v1/organizers/europython/events/{event}/"
+    return url
+
+
+def fetch_pretix_data(
+    event: str, resource: PretixData.PretixResources
+) -> list[JsonType]:
+    headers = {
+        "Authorization": f"Token {settings.PRETIX_API_TOKEN}",
+        "Content-Type": "application/json",
+    }
+
+    base_url = get_event_url(event)
+    endpoint = ENDPOINTS[resource]
+    url = f"{base_url}{endpoint}"
+
+    # Pretalx paginates the output, so we will need to do multiple requests and
+    # then merge multiple pages to one big dictionary
+    results = []
+    page = 0
+
+    # This takes advantage of the fact that url will contain a url to the
+    # next page, until there is more data to fetch. If this is the last page,
+    # then the url will be None (falsy), and thus stop the while loop.
+    while url:
+        page += 1
+        response = httpx.get(url, headers=headers)
+
+        if response.status_code != 200:
+            raise Exception(f"Error {response.status_code}: {response.text}")
+
+        logger.info("Fetching data from %s, page %s", url, page)
+
+        data = response.json()
+        results += data["results"]
+        url = data["next"]
+
+    return results
+
+
+def download_latest_orders(event: str) -> PretixData:
+    data = fetch_pretix_data(event, PretixData.PretixResources.orders)
+
+    pretalx_data = PretixData.objects.create(
+        resource=PretixData.PretixResources.orders,
+        content=data,
+    )
+
+    return pretalx_data

intbot/core/models.py

@@ -87,9 +87,7 @@ 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".
+    We first download data from pretix, then we later parse the latest jsons.
     """
 
     class PretalxResources(models.TextChoices):
@@ -110,3 +108,30 @@ class PretalxResources(models.TextChoices):
 
     def __str__(self):
         return f"{self.uuid}"
+
+
+class PretixData(models.Model):
+    """
+    Table to store raw data download from pretix for later parsing.
+
+    We first download data from pretix, then we later parse the latest jsons.
+    """
+
+    class PretixResources(models.TextChoices):
+        orders = "orders", "Orders"
+        products = "products", "Products"
+        vouchers = "vouchers", "Vouchers"
+
+    uuid = models.UUIDField(default=uuid.uuid4)
+    resource = models.CharField(
+        max_length=255,
+        choices=PretixResources.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

@@ -200,6 +200,9 @@ def get(name) -> str:
 # Pretalx
 PRETALX_API_TOKEN = get("PRETALX_API_TOKEN")
 
+# Pretix
+PRETIX_API_TOKEN = get("PRETIX_API_TOKEN")
+
 
 if DJANGO_ENV == "dev":
     DEBUG = True

intbot/tests/test_integrations/test_pretix.py

@@ -0,0 +1,99 @@
+import pytest
+import respx
+from core.integrations import pretix
+from core.models import PretixData
+from httpx import Response
+
+
+def orders_pages_generator(url):
+    """
+    Generator to simulate pagination.
+
+    Extracted to a generator because we use it in multiple places
+    """
+    yield Response(
+        200,
+        json={
+            "results": [
+                {"hello": "world"},
+            ],
+            "next": f"{url}&page=2",
+        },
+    )
+
+    yield Response(
+        200,
+        json={
+            "results": [
+                {"foo": "bar"},
+            ],
+            # It's important to make it last page in tests.
+            # Otherwise it will be infinite loop :)
+            "next": None,
+        },
+    )
+
+
+def speaker_pages_generator(url):
+    """
+    Generator to simulate pagination.
+
+    Extracted to a generator because we use it in multiple places
+    """
+    yield Response(
+        200,
+        json={
+            "results": [
+                {"hello": "world"},
+            ],
+            "next": f"{url}&page=2",
+        },
+    )
+
+    yield Response(
+        200,
+        json={
+            "results": [
+                {"foo": "bar"},
+            ],
+            # It's important to make it last page in tests.
+            # Otherwise it will be infinite loop :)
+            "next": None,
+        },
+    )
+
+
+@respx.mock
+def test_fetch_orders_from_pretix():
+    url = "https://tickets.europython.eu/api/v1/organizers/europython/ep2025/orders/"
+    data = orders_pages_generator(url)
+    respx.get(url).mock(return_value=next(data))
+    respx.get(url + "&page=2").mock(return_value=next(data))
+
+    orders = pretix.fetch_pretix_data(
+        "ep2025",
+        PretixData.PretixResources.orders,
+    )
+
+    assert orders == [
+        {"hello": "world"},
+        {"foo": "bar"},
+    ]
+
+
+@respx.mock
+@pytest.mark.django_db
+def test_download_latest_orders():
+    url = "https://tickets.europython.eu/api/v1/organizers/europython/ep2025/orders/"
+    data = orders_pages_generator(url)
+    respx.get(url).mock(return_value=next(data))
+    respx.get(url + "&page=2").mock(return_value=next(data))
+
+    pretix.download_latest_orders("ep2025")
+
+    pd = PretixData.objects.get(resource=PretixData.PretixResources.orders)
+    assert pd.resource == "orders"
+    assert pd.content == [
+        {"hello": "world"},
+        {"foo": "bar"},
+    ]

pyproject.toml

@@ -25,6 +25,7 @@ dependencies = [
     "respx>=0.22.0",
     "pydantic>=2.10.6",
     "freezegun>=1.5.1",
+    "ipython>=9.1.0",
 ]
 
 [tool.pytest.ini_options]