Eksporter til et datavarehus

Analytikere vil ofte ha Sursa-data ved siden av resten av bedriftens data: hvilke kort som er verifisert og av hvem, hvordan kunnskapen vokser per team, hvilke spørsmål som forblir ubesvart. Denne veiledningen bygger en planlagt jobb som kopierer Sursa til datavarehuset ditt og holder kopien oppdatert.

Jobben kombinerer tre deler av API-et:

Du trenger en nøkkel med scopene cards:read og users:read, pluss webhooks:manage for hendelsesfeeden. Gi jobben sin egen nøkkel med rollen admin: Den fullstendige listen over brukere og hendelsesfeeden er bare åpne for den. En nøkkel med en smalere rolle og dataprofil eksporterer bare det en bestemt gruppe har lov til å se; den leser den korte listen over brukere (brief=true) og ingen hendelsesfeed, så den får slettingene fra den ukentlige fulle innlastingen nedenfor.

Tabeller

Et enkelt oppsett med én tabell per objekt og én tabell for hendelser:

Lagre hvert objekt slik det kommer, som en JSON-kolonne, pluss kolonnene du ofte spør etter (updated_at, collection_id, status, owner_id). Nye felt dukker opp i senere versjoner av API-et; med JSON-kolonnen får du dem uten å endre jobben.

Første innlasting

Les hver liste fullstendig, uten updated_since, og skriv alle objektene. Husk tidspunktet da innlastingen startet: Neste kjøring fortsetter derfra.

Inkrementelle innlastinger

Ved hver kjøring, for eksempel hvert 15. minutt:

1. Husk starttidspunktet for denne kjøringen.
2. Les kortene med updated_since satt til starttidspunktet for forrige kjøring, minus ett minutts margin, og be om hver neste side med cursoren og den samme updated_since (en cursor uten filtrene fra den første siden returnerer 400 api.cursor_invalid). Les samlingene, brukerne og gruppene i sin helhet: Samlinger og brukere har ikke updated_at, og updated_at for en gruppe endres ikke når medlemmene endres. Opprett eller erstatt hvert objekt etter id.
3. Les hendelsesfeeden fra cursoren du lagret sist. Legg til hver hendelse i sursa_events, og utfør slettingene: En card.deleted-hendelse fjerner kortet fra sursa_cards eller merker det som slettet hvis du tar vare på historikken.
4. Lagre starttidspunktet og den nye cursoren for feeden, i samme transaksjon som dataene hvis datavarehuset ditt stø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})

Feedens cursor returneres på hver side, også den siste. Når du lagrer den, starter neste kjøring rett etter den siste hendelsen du lagret, så ingen hendelse leses to ganger eller hoppes over.

Hold kopien korrekt

• Mer enn 30 dager uten kjøring. Feeden tar vare på hendelser i 30 dager. Hvis jobben ikke har kjørt på lengre tid, kjører du en full innlasting på nytt og fjerner fra tabellene alle objekter som den fulle innlastingen ikke returnerte.
• Ukentlig full innlasting. Selv med en feed som fungerer som den skal, er en full innlasting én gang i uken, sammenlignet med tabellene dine, en billig kontroll av at kopien er komplett.
• Det nøkkelen ikke kan se, eksporteres ikke. Jobben ser nøyaktig det nøkkelen ser. Hvis et kort havner utenfor nøkkelens tilgang, for eksempel fordi det ble flyttet til en begrenset samling, returnerer en full innlasting det ikke; behandle det som fjernet.
• Spørsmål. Spørsmålsloggen har sin egen eksport: GET /v1/questions/export.csv returnerer spørsmålene fra en periode som én CSV-fil, med vurderinger og kilder. Den krever scopet ask.

Respekter grensene

En full innlasting for en stor organisasjon sender mange forespørsler. Les én liste om gangen med limit=200, og følg med på RateLimit-headerne. En jobb hvert 15. minutt som bare leser endringer, holder seg langt under alle kvoter.