Esportare in un data warehouse

Gli analisti vogliono spesso i dati di Sursa accanto al resto dei dati aziendali: quali schede sono verificate e da chi, come cresce la conoscenza per team, quali domande restano senza risposta. Questa guida costruisce un job pianificato che copia Sursa nel tuo warehouse e mantiene la copia aggiornata.

Il job combina tre parti dell’API:

Ti serve una chiave con gli ambiti cards:read e users:read, più webhooks:manage per il feed degli eventi. Dai al job una chiave propria con il ruolo admin: l’elenco completo degli utenti e il feed degli eventi sono accessibili solo a una chiave di questo tipo. Una chiave con un ruolo e un profilo dati più ristretti esporta solo ciò che un determinato gruppo può vedere; legge l’elenco sintetico degli utenti (brief=true) e nessun feed degli eventi, quindi ricava le eliminazioni dal caricamento completo settimanale descritto più avanti.

Tabelle

Una struttura semplice, con una tabella per ogni oggetto e una tabella per gli eventi:

Conserva ogni oggetto così come arriva, in una colonna JSON, più le colonne che interroghi spesso (updated_at, collection_id, status, owner_id). Nelle versioni successive dell’API compaiono nuovi campi; con la colonna JSON li ricevi senza modificare il job.

Primo caricamento

Leggi ogni elenco per intero, senza updated_since, e scrivi ogni oggetto. Annota il momento in cui il caricamento è iniziato: l’esecuzione successiva riparte da lì.

Caricamenti incrementali

A ogni esecuzione, per esempio ogni 15 minuti:

1. Annota l’orario di inizio di questa esecuzione.
2. Leggi le schede con updated_since impostato sull’orario di inizio dell’esecuzione precedente, meno un minuto di margine, e richiedi ogni pagina successiva con il cursore e lo stesso updated_since (un cursore senza i filtri della prima pagina restituisce 400 api.cursor_invalid). Leggi per intero raccolte, utenti e gruppi: raccolte e utenti non hanno updated_at, e l’updated_at di un gruppo non cambia quando cambiano i suoi membri. Inserisci o aggiorna ogni oggetto in base all’id.
3. Leggi il feed degli eventi dal cursore che hai salvato l’ultima volta. Aggiungi ogni evento a sursa_events e applica le eliminazioni: un evento card.deleted rimuove la scheda da sursa_cards, oppure la contrassegna come eliminata se conservi la cronologia.
4. Salva l’orario di inizio e il nuovo cursore del feed, nella stessa transazione dei dati se il tuo warehouse lo consente.

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

Il cursore del feed viene restituito in ogni pagina, anche nell’ultima. Salvarlo significa che l’esecuzione successiva inizia esattamente dopo l’ultimo evento che hai salvato, così nessun evento viene letto due volte o saltato.

Mantenere fedele la copia

• Più di 30 giorni senza esecuzioni. Il feed conserva gli eventi per 30 giorni. Se il job non è stato eseguito per più tempo, esegui di nuovo un caricamento completo e rimuovi dalle tue tabelle ogni oggetto che il caricamento completo non ha restituito.
• Caricamento completo settimanale. Anche con un feed in buono stato, un caricamento completo una volta alla settimana, confrontato con le tue tabelle, è un controllo economico che la copia sia completa.
• Ciò che la chiave non vede non viene esportato. Il job vede esattamente ciò che vede la sua chiave. Se una scheda esce dall’accesso della chiave, per esempio perché è stata spostata in una raccolta riservata, un caricamento completo non la restituisce; trattala come rimossa.
• Domande. Il registro delle domande ha una propria esportazione: GET /v1/questions/export.csv restituisce le domande di un periodo come un unico file CSV, con valutazioni e fonti. Richiede l’ambito ask.

Rispettare i limiti

Un caricamento completo di una grande organizzazione fa molte richieste. Leggi un elenco alla volta con limit=200 e tieni d’occhio le intestazioni RateLimit. Un job ogni 15 minuti che legge solo le modifiche resta molto al di sotto di qualsiasi quota.