Eksportér til et datavarehus

Analytikere vil ofte have Sursas data ved siden af virksomhedens øvrige data: hvilke kort der er verificeret og af hvem, hvordan viden vokser pr. team, hvilke spørgsmål der forbliver ubesvarede. I denne vejledning bygger du et planlagt job, der kopierer Sursa til dit datavarehus og holder kopien opdateret.

Jobbet kombinerer tre dele af API’et:

Du skal bruge en nøgle med scopes cards:read og users:read samt webhooks:manage til hændelsesfeedet. Giv jobbet sin egen nøgle med rollen admin: Den fulde liste over brugere og hændelsesfeedet er kun åbne for den. En nøgle med en snævrere rolle og dataprofil eksporterer kun det, en bestemt gruppe må se; den læser den korte liste over brugere (brief=true) og intet hændelsesfeed, så den henter sletninger fra den ugentlige fulde indlæsning nedenfor.

Tabeller

Et enkelt layout med én tabel pr. objekt og én tabel til hændelser:

Gem hvert objekt, som det kommer, i en JSON-kolonne, samt de kolonner, du ofte forespørger på (updated_at, collection_id, status, owner_id). Nye felter dukker op i senere versioner af API’et; med JSON-kolonnen får du dem uden at ændre jobbet.

Første indlæsning

Læs hver liste helt, uden updated_since, og skriv hvert objekt. Husk det tidspunkt, hvor indlæsningen startede: Den næste kørsel fortsætter derfra.

Inkrementelle indlæsninger

Ved hver kørsel, fx hvert 15. minut:

1. Husk starttidspunktet for denne kørsel.
2. Læs kortene med updated_since sat til starttidspunktet for den forrige kørsel minus et minuts margin, og hent hver følgende side med cursoren og det samme updated_since (en cursor uden filtrene fra den første side returnerer 400 api.cursor_invalid). Læs samlinger, brugere og grupper hele: Samlinger og brugere har intet updated_at, og en gruppes updated_at ændrer sig ikke, når dens medlemmer ændrer sig. Opret eller erstat hvert objekt efter id.
3. Læs hændelsesfeedet fra den cursor, du gemte sidst. Føj hver hændelse til sursa_events, og udfør sletningerne: En card.deleted-hændelse fjerner kortet fra sursa_cards eller markerer det som slettet, hvis du gemmer historik.
4. Gem starttidspunktet og den nye cursor til feedet i samme transaktion som dataene, hvis dit datavarehus understøtter det.

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

Feedets cursor returneres på hver side, også den sidste. Når du gemmer den, starter den næste kørsel præcis efter den sidste hændelse, du har gemt, så ingen hændelse læses to gange eller springes over.

Hold kopien retvisende

• Mere end 30 dage uden kørsel. Feedet gemmer hændelser i 30 dage. Har jobbet ikke kørt i længere tid, så kør en fuld indlæsning igen, og fjern alle objekter fra dine tabeller, som den fulde indlæsning ikke returnerede.
• Ugentlig fuld indlæsning. Selv med et sundt feed er en fuld indlæsning én gang om ugen, sammenlignet med dine tabeller, en billig kontrol af, at kopien er komplet.
• Det, nøglen ikke kan se, eksporteres ikke. Jobbet ser nøjagtig det, som dets nøgle ser. Forsvinder et kort fra nøglens adgang, fx fordi det er flyttet til en begrænset samling, returnerer en fuld indlæsning det ikke; behandl det som fjernet.
• Spørgsmål. Spørgsmålsloggen har sin egen eksport: GET /v1/questions/export.csv returnerer spørgsmålene for en periode som én CSV-fil med vurderinger og kilder. Den kræver scopet ask.

Overhold grænserne

En fuld indlæsning af en stor organisation sender mange anmodninger. Læs én liste ad gangen med limit=200, og hold øje med RateLimit-headerne. Et job hvert 15. minut, der kun læser ændringer, holder sig langt under enhver kvote.