feat(website): python seeder with pango lineages and test suite

.github/workflows/example-data-seeder.yml

@@ -6,8 +6,24 @@ env:
   DOCKER_IMAGE_NAME: ghcr.io/genspectrum/dashboards/example-data-seeder
 
 jobs:
+  test:
+    name: Test Example Data Seeder
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+
+      - uses: prefix-dev/setup-pixi@v0.9.6
+        with:
+          manifest-path: collection-seeding/pixi.toml
+          environments: test
+
+      - name: Run tests
+        working-directory: ./collection-seeding
+        run: pixi run -e test test
+
   dockerImage:
     name: Build Example Data Seeder Docker Image
+    needs: test
     runs-on: ubuntu-latest
     permissions:
       contents: read
@@ -37,7 +53,7 @@ jobs:
       - name: Build and push image
         uses: docker/build-push-action@v7
         with:
-          context: ./example-data
+          context: ./collection-seeding
           tags: ${{ steps.dockerMetadata.outputs.tags }}
           cache-from: type=gha,scope=example-data-seeder-${{ github.ref }}
           cache-to: type=gha,mode=max,scope=example-data-seeder-${{ github.ref }}

.gitignore

@@ -13,3 +13,10 @@ logs
 node_modules/
 
 .env
+
+# Python
+__pycache__/
+*.pyc
+
+# pixi
+.pixi/

collection-seeding/.dockerignore

@@ -0,0 +1,7 @@
+*
+!pixi.toml
+!pixi.lock
+!seed.py
+!api.py
+!models.py
+!sources/

collection-seeding/Dockerfile

@@ -0,0 +1,13 @@
+# Stage 1: use pixi to resolve and install dependencies
+FROM ghcr.io/prefix-dev/pixi:0.58.0 AS builder
+WORKDIR /app
+COPY pixi.toml pixi.lock .
+RUN pixi install --frozen
+
+# Stage 2: slim runtime image — copy only the installed site-packages
+FROM python:3.13-slim AS final
+WORKDIR /app
+COPY --from=builder /app/.pixi/envs/default/lib/python3.13/site-packages \
+                    /usr/local/lib/python3.13/site-packages
+COPY . .
+CMD ["python", "seed.py"]

collection-seeding/README.md

@@ -0,0 +1,59 @@
+# collection-seeding
+
+Seeds the backend with example collections:
+
+- **covid-resistance-mutations** — resistance mutation data for 3CLpro, RdRp, and Spike mAb
+- **covid-pango-lineages** — one collection per pango lineage, with nucleotide substitutions as variants
+- **covid-pango-lineages-sample** — same as above but limited to 10 lineages, for quick testing
+
+The script is idempotent — re-running it will create new collections or update existing ones (matched by name). If a collection's name changes in the source, the old entry is orphaned and a new one is created.
+
+Use `--repeat-interval-hours N` (or `$REPEAT_INTERVAL_HOURS`) to run on a loop — re-seeds every N hours. Without it, the script runs once and exits.
+
+## Via Docker Compose
+
+The seeder runs automatically as part of Docker Compose:
+
+```bash
+BACKEND_TAG=latest WEBSITE_TAG=latest SEEDER_TAG=latest docker compose up
+```
+
+## Running locally
+
+Requires [pixi](https://pixi.sh). Install dependencies once:
+
+```bash
+pixi install
+```
+
+Then use the provided tasks:
+
+```bash
+pixi run seed                    # all sources
+pixi run seed-resistance         # resistance mutations only
+pixi run seed-lineages           # pango lineages only
+pixi run seed-lineages-sample    # first 10 pango lineages (quick test)
+```
+
+To target a different backend:
+
+```bash
+pixi run seed --url http://localhost:4321
+```
+
+Run `pixi run seed --help` for all options, including `--source`, `--list`, `--repeat-interval-hours`, and `--url`.
+
+## Adding a new source
+
+1. Create `sources/your_source.py` and implement the `Source` ABC:
+   ```python
+   from sources import Source
+   from models import Collection
+
+   class YourSource(Source):
+       name = "your-source-name"  # used with --source flag
+
+       def get_collections(self) -> list[Collection]:
+           ...
+   ```
+2. Register it in `sources/registry.py` by adding it to `ALL_SOURCES`.

collection-seeding/api.py

@@ -0,0 +1,83 @@
+"""Shared backend API client for collection seeders."""
+
+import time
+
+import requests
+
+from models import Collection, ExistingCollection
+
+RETRY_ATTEMPTS = 30
+RETRY_DELAY_S = 2
+
+
+class ApiClient:
+    def __init__(self, base_url: str, api_key: str):
+        self.base_url = base_url.rstrip("/")
+        self._collections_url = f"{self.base_url}/api/collections"
+        self._auth_headers = {"Authorization": f"Bearer {api_key}"}
+
+    def wait_for_api(
+        self, attempts: int = RETRY_ATTEMPTS, delay: float = RETRY_DELAY_S
+    ):
+        """Poll until the API is ready by checking the collections endpoint."""
+        for attempt in range(1, attempts + 1):
+            try:
+                r = requests.get(self._collections_url, timeout=10)
+                if r.ok:
+                    return
+            except requests.RequestException:
+                pass
+            print(f"Waiting for API... (attempt {attempt}/{attempts})")
+            time.sleep(delay)
+        raise RuntimeError(
+            f"API at {self.base_url} did not become ready after {attempts} attempts."
+        )
+
+    def fetch_existing_collections(self, organism: str) -> list[ExistingCollection]:
+        r = requests.get(
+            self._collections_url,
+            params={"organism": organism},
+            headers=self._auth_headers,
+            timeout=10,
+        )
+        if not r.ok:
+            raise RuntimeError(f"GET /api/collections failed: {r.status_code} {r.text}")
+        return r.json()
+
+    def create_collection(self, collection: Collection) -> int:
+        r = requests.post(
+            self._collections_url,
+            headers=self._auth_headers,
+            json=collection,
+            timeout=10,
+        )
+        if r.status_code != 201:
+            raise RuntimeError(
+                f"POST /api/collections failed: {r.status_code} {r.text}"
+            )
+        return r.json()["id"]
+
+    def delete_collection(self, collection_id: int) -> None:
+        r = requests.delete(
+            f"{self._collections_url}/{collection_id}",
+            headers=self._auth_headers,
+            timeout=10,
+        )
+        if not r.ok:
+            raise RuntimeError(
+                f"DELETE /api/collections/{collection_id} failed: {r.status_code} {r.text}"
+            )
+
+    def update_collection(self, collection_id: int, collection: Collection) -> None:
+        # CollectionUpdate has no organism field; sending it causes a 400 (fail-on-unknown-properties=true)
+        body = {k: v for k, v in collection.items() if k != "organism"}
+        r = requests.put(
+            f"{self._collections_url}/{collection_id}",
+            headers=self._auth_headers,
+            json=body,
+            timeout=10,
+        )
+        if not r.ok:
+            raise RuntimeError(
+                f"PUT /api/collections/{collection_id} failed: {r.status_code} {r.text}"
+            )

collection-seeding/models.py

@@ -0,0 +1,29 @@
+"""Shared type definitions for collection seeding."""
+
+from typing import TypedDict
+
+
+class FilterObject(TypedDict, total=False):
+    aminoAcidMutations: list[str]
+    nucleotideMutations: list[str]
+
+
+class Variant(TypedDict):
+    type: str
+    name: str
+    filterObject: FilterObject
+
+
+class Collection(TypedDict):
+    name: str
+    organism: str
+    description: str
+    variants: list[Variant]
+
+
+class ExistingCollection(TypedDict):
+    """A collection as returned by the backend (includes the assigned id)."""
+
+    id: int
+    name: str
+    description: str | None