Exportul într-un depozit de date

Analiștii vor adesea datele din Sursa alături de restul datelor companiei: ce fișe sunt verificate și de cine, cum cresc cunoștințele fiecărei echipe, ce întrebări rămân fără răspuns. Acest ghid construiește un job programat care copiază datele din Sursa în depozitul de date și păstrează copia la zi.

Jobul combină trei părți ale API-ului:

Aveți nevoie de o cheie cu domeniile cards:read și users:read, plus webhooks:manage pentru fluxul de evenimente. Dați jobului propria cheie, cu rolul admin: lista completă a utilizatorilor și fluxul de evenimente sunt accesibile doar acestui rol. O cheie cu un rol și un profil de date mai restrânse exportă doar ce poate vedea un anumit grup; ea citește lista scurtă a utilizatorilor (brief=true) și nu are acces la fluxul de evenimente, așa că preia ștergerile din preluarea completă săptămânală descrisă mai jos.

Tabele

O structură simplă, cu un tabel pentru fiecare obiect și un tabel pentru evenimente:

Stocați fiecare obiect așa cum vine, într-o coloană JSON, plus coloanele pe care le interogați des (updated_at, collection_id, status, owner_id). În versiunile ulterioare ale API-ului apar câmpuri noi; cu coloana JSON le primiți fără să modificați jobul.

Preluarea inițială

Citiți complet fiecare listă, fără updated_since, și scrieți fiecare obiect. Rețineți momentul în care preluarea a început: rularea următoare continuă de acolo.

Preluări incrementale

La fiecare rulare, de exemplu la fiecare 15 minute:

1. Rețineți momentul de început al acestei rulări.
2. Citiți fișele cu updated_since setat la momentul de început al rulării anterioare, minus un minut de marjă, și cereți fiecare pagină următoare cu cursorul și același updated_since (un cursor fără filtrele primei pagini returnează 400 api.cursor_invalid). Citiți integral colecțiile, utilizatorii și grupurile: colecțiile și utilizatorii nu au updated_at, iar updated_at al unui grup nu se schimbă când se schimbă membrii lui. Faceți creare sau înlocuire (upsert) pentru fiecare obiect, după id.
3. Citiți fluxul de evenimente de la cursorul salvat data trecută. Adăugați fiecare eveniment în sursa_events și aplicați ștergerile: un eveniment card.deleted elimină fișa din sursa_cards sau o marchează ca ștearsă, dacă păstrați istoricul.
4. Salvați momentul de început și noul cursor al fluxului, în aceeași tranzacție cu datele, dacă depozitul permite acest lucru.

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

Cursorul fluxului este returnat pe fiecare pagină, chiar și pe ultima. Dacă îl salvați, rularea următoare începe exact după ultimul eveniment stocat, așa că niciun eveniment nu este citit de două ori sau sărit.

O copie fidelă

• Peste 30 de zile fără rulare. Fluxul păstrează evenimentele 30 de zile. Dacă jobul nu a rulat mai mult de atât, faceți din nou o preluare completă și eliminați din tabele fiecare obiect pe care preluarea completă nu l-a returnat.
• Preluare completă săptămânală. Chiar și cu un flux în bună stare, o preluare completă o dată pe săptămână, comparată cu tabelele, este o verificare ieftină a faptului că copia este completă.
• Ce nu vede cheia nu se exportă. Jobul vede exact ce vede cheia lui. Dacă o fișă iese din accesul cheii, de exemplu pentru că a fost mutată într-o colecție restricționată, o preluare completă nu o va returna; tratați-o ca eliminată.
• Întrebările. Jurnalul întrebărilor are propriul export: GET /v1/questions/export.csv returnează întrebările dintr-o perioadă ca un singur fișier CSV, cu evaluări și surse. Necesită domeniul ask.

Respectați limitele

O preluare completă a unei organizații mari face multe cereri. Citiți câte o listă pe rând, cu limit=200, și urmăriți antetele RateLimit. Un job la fiecare 15 minute care citește doar modificările rămâne mult sub orice cotă.