Export in ein Data Warehouse

Analystinnen und Analysten möchten Sursa-Daten oft neben den übrigen Daten des Unternehmens haben: welche Karten geprüft sind und von wem, wie das Wissen pro Team wächst, welche Fragen unbeantwortet bleiben. Diese Anleitung baut einen geplanten Job, der Sursa in Ihr Warehouse kopiert und die Kopie aktuell hält.

Der Job verbindet drei Teile der API:

Sie brauchen einen Schlüssel mit den Scopes cards:read und users:read sowie webhooks:manage für den Ereignis-Feed. Geben Sie dem Job einen eigenen Schlüssel mit der Rolle admin: Nur einem solchen Schlüssel stehen die vollständige Liste der Benutzer und der Ereignis-Feed offen. Ein Schlüssel mit einer engeren Rolle und einem Datenprofil exportiert nur, was eine bestimmte Gruppe sehen darf; er liest die Kurzliste der Benutzer (brief=true) und keinen Ereignis-Feed und übernimmt Löschungen daher aus der wöchentlichen vollständigen Ladung unten.

Tabellen

Ein einfaches Schema mit einer Tabelle pro Objekt und einer Tabelle für Ereignisse:

Speichern Sie jedes Objekt so, wie es kommt, als JSON-Spalte, dazu die Spalten, die Sie häufig abfragen (updated_at, collection_id, status, owner_id). In späteren Versionen der API kommen neue Felder hinzu; mit der JSON-Spalte erhalten Sie sie, ohne den Job zu ändern.

Erste Ladung

Lesen Sie jede Liste vollständig, ohne updated_since, und schreiben Sie jedes Objekt. Merken Sie sich den Zeitpunkt, an dem die Ladung begonnen hat: Der nächste Lauf macht dort weiter.

Inkrementelle Ladungen

Bei jedem Lauf, zum Beispiel alle 15 Minuten:

1. Merken Sie sich die Startzeit dieses Laufs.
2. Lesen Sie die Karten mit updated_since gleich der Startzeit des vorherigen Laufs abzüglich einer Minute Reserve, und fordern Sie jede weitere Seite mit dem Cursor und demselben updated_since an (ein Cursor ohne die Filter der ersten Seite ergibt 400 api.cursor_invalid). Lesen Sie Sammlungen, Benutzer und Gruppen vollständig: Sammlungen und Benutzer haben kein updated_at, und das updated_at einer Gruppe ändert sich nicht, wenn sich ihre Mitglieder ändern. Schreiben Sie jedes Objekt per Upsert anhand von id.
3. Lesen Sie den Ereignis-Feed ab dem Cursor, den Sie beim letzten Mal gespeichert haben. Hängen Sie jedes Ereignis an sursa_events an und wenden Sie die Löschungen an: Ein Ereignis card.deleted entfernt die Karte aus sursa_cards oder markiert sie als gelöscht, wenn Sie einen Verlauf führen.
4. Speichern Sie die Startzeit und den neuen Feed-Cursor, wenn Ihr Warehouse es unterstützt, in derselben Transaktion wie die Daten.

import json
from datetime import datetime, timedelta, timezone

LISTS = {"cards": "/v1/cards", "collections": "/v1/collections", "users": "/v1/users", "groups": "/v1/groups"}


def read_list(path, since):
    """Every object of a list, changed since `since` (or all of them). The next page keeps the filters of the first."""
    params = {"limit": 200}
    if since:
        params["updated_since"] = (since - timedelta(minutes=1)).isoformat()
    while params:
        response = session.get(BASE + path, params=params, timeout=120)
        response.raise_for_status()
        body = response.json()
        items = body["data"] if isinstance(body, dict) and "data" in body else body
        yield from items
        if isinstance(body, dict) and "data" in body:
            params = {**params, "cursor": body["next_cursor"]} if body.get("has_more") else None
        else:
            cursor = response.headers.get("X-Next-Cursor")
            params = {**params, "cursor": cursor} if cursor else None


def read_events(cursor):
    """New events of the feed after `cursor` (or from the start of the 30 days), and the cursor to save."""
    events = []
    params = {"limit": 200, **({"cursor": cursor} if cursor else {})}
    while True:
        page = api("GET", "/v1/webhook-events?" + requests.compat.urlencode(params))
        events += page["data"]
        params["cursor"] = page["next_cursor"]
        if not page["has_more"]:
            return events, page["next_cursor"]


def run(state, warehouse):
    started_at = datetime.now(timezone.utc)
    since = state.get("since")
    for table, path in LISTS.items():
        for obj in read_list(path, since if table == "cards" else None):
            warehouse.upsert(f"sursa_{table}", obj["id"], json.dumps(obj))

    events, feed_cursor = read_events(state.get("feed_cursor"))
    for event in events:
        warehouse.insert_once("sursa_events", event["id"], json.dumps(event))
        if event["type"] == "card.deleted":
            warehouse.delete("sursa_cards", event["object"]["id"])

    warehouse.save_state({"since": started_at, "feed_cursor": feed_cursor})

Der Feed-Cursor wird auf jeder Seite zurückgegeben, auch auf der letzten. Wenn Sie ihn speichern, beginnt der nächste Lauf genau nach dem letzten gespeicherten Ereignis, sodass kein Ereignis doppelt gelesen oder übersprungen wird.

Die Kopie verlässlich halten

• Mehr als 30 Tage ohne Lauf. Der Feed bewahrt Ereignisse 30 Tage auf. Ist der Job länger nicht gelaufen, führen Sie erneut eine vollständige Ladung aus und entfernen Sie aus Ihren Tabellen jedes Objekt, das die vollständige Ladung nicht zurückgegeben hat.
• Wöchentliche vollständige Ladung. Auch bei einem fehlerfreien Feed ist eine vollständige Ladung einmal pro Woche, verglichen mit Ihren Tabellen, eine günstige Prüfung, ob die Kopie vollständig ist.
• Was der Schlüssel nicht sieht, wird nicht exportiert. Der Job sieht genau das, was sein Schlüssel sieht. Verlässt eine Karte den Zugriffsbereich des Schlüssels, etwa weil sie in eine eingeschränkte Sammlung verschoben wurde, gibt eine vollständige Ladung sie nicht zurück; behandeln Sie sie als entfernt.
• Fragen. Das Fragenprotokoll hat einen eigenen Export: GET /v1/questions/export.csv gibt die Fragen eines Zeitraums als eine CSV-Datei zurück, mit Bewertungen und Quellen. Dafür ist der Scope ask nötig.

Limits einhalten

Eine vollständige Ladung einer großen Organisation stellt viele Anfragen. Lesen Sie jeweils eine Liste mit limit=200 und beobachten Sie die RateLimit-Header. Ein Job alle 15 Minuten, der nur Änderungen liest, bleibt weit unter jedem Kontingent.