Export naar een datawarehouse

Analisten willen de gegevens van Sursa vaak naast de rest van de bedrijfsgegevens hebben: welke kaarten zijn geverifieerd en door wie, hoe groeit de kennis per team, welke vragen blijven onbeantwoord. In deze handleiding bouw je een geplande job die Sursa naar je datawarehouse kopieert en de kopie actueel houdt.

De job combineert drie onderdelen van de API:

Je hebt een sleutel nodig met de scopes cards:read en users:read, plus webhooks:manage voor de eventfeed. Geef de job een eigen sleutel met de rol admin: alleen zo’n sleutel heeft toegang tot de volledige lijst van gebruikers en tot de eventfeed. Een sleutel met een beperktere rol en een dataprofiel exporteert alleen wat een bepaalde groep mag zien; hij leest de beknopte lijst van gebruikers (brief=true) en geen eventfeed, dus hij haalt verwijderingen uit de wekelijkse volledige lading hieronder.

Tabellen

Een eenvoudige indeling met één tabel per object en één tabel voor events:

Sla elk object op zoals het binnenkomt, als JSON-kolom, plus de kolommen waarop je vaak zoekt (updated_at, collection_id, status, owner_id). In latere versies van de API komen er nieuwe velden bij; met de JSON-kolom krijg je die zonder de job aan te passen.

Eerste lading

Lees elke lijst volledig, zonder updated_since, en schrijf elk object weg. Onthoud het moment waarop de lading begon: de volgende run gaat vanaf daar verder.

Incrementele ladingen

Bij elke run, bijvoorbeeld elke 15 minuten:

1. Onthoud het begintijdstip van deze run.
2. Lees de kaarten met updated_since ingesteld op het begintijdstip van de vorige run, min een minuut marge, en vraag elke volgende pagina op met de cursor en dezelfde updated_since (een cursor zonder de filters van de eerste pagina geeft 400 api.cursor_invalid). Lees de collecties, gebruikers en groepen in hun geheel: collecties en gebruikers hebben geen updated_at, en de updated_at van een groep verandert niet als de leden ervan veranderen. Maak of vervang elk object op basis van id (upsert).
3. Lees de eventfeed vanaf de cursor die je de vorige keer hebt bewaard. Voeg elk event toe aan sursa_events en verwerk de verwijderingen: een event card.deleted verwijdert de kaart uit sursa_cards, of markeert hem als verwijderd als je de geschiedenis bewaart.
4. Bewaar het begintijdstip en de nieuwe cursor van de feed, in dezelfde transactie als de gegevens als je datawarehouse dat ondersteunt.

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})

De cursor van de feed wordt op elke pagina teruggegeven, ook op de laatste. Door hem te bewaren begint de volgende run precies na het laatste event dat je hebt opgeslagen, zodat geen enkel event twee keer wordt gelezen of wordt overgeslagen.

De kopie betrouwbaar houden

• Meer dan 30 dagen zonder run. De feed bewaart events 30 dagen. Heeft de job langer niet gedraaid, voer dan opnieuw een volledige lading uit en verwijder uit je tabellen elk object dat de volledige lading niet heeft teruggegeven.
• Wekelijkse volledige lading. Ook met een gezonde feed is een volledige lading eens per week, vergeleken met je tabellen, een goedkope controle of de kopie compleet is.
• Wat de sleutel niet kan zien, wordt niet geëxporteerd. De job ziet precies wat zijn sleutel ziet. Valt een kaart buiten de toegang van de sleutel, bijvoorbeeld omdat hij naar een afgeschermde collectie is verplaatst, dan geeft een volledige lading hem niet terug; behandel hem als verwijderd.
• Vragen. Het vragenlog heeft een eigen export: GET /v1/questions/export.csv geeft de vragen van een periode terug als één CSV-bestand, met beoordelingen en bronnen. Daarvoor is de scope ask nodig.

Binnen de limieten blijven

Een volledige lading van een grote organisatie doet veel requests. Lees één lijst tegelijk met limit=200 en let op de RateLimit-headers. Een job die elke 15 minuten alleen wijzigingen leest, blijft ver onder elk quotum.