Vienti tietovarastoon

Analyytikot haluavat usein Sursan tiedot yrityksen muiden tietojen rinnalle: mitkä kortit on vahvistettu ja kuka ne vahvisti, miten tieto karttuu tiimeittäin, mitkä kysymykset jäävät vaille vastausta. Tässä oppaassa rakennetaan ajastettu työ, joka kopioi Sursan tiedot tietovarastoosi ja pitää kopion ajan tasalla.

Työ yhdistää kolme API:n osaa:

Tarvitset avaimen, jolla on käyttöalueet cards:read ja users:read sekä tapahtumasyötettä varten webhooks:manage. Anna työlle oma avain, jolla on rooli admin: käyttäjien täysi lista ja tapahtumasyöte ovat avoinna vain sille. Avain, jolla on suppeampi rooli ja dataprofiili, vie vain sen, minkä tietty ryhmä saa nähdä; se lukee käyttäjien suppean listan (brief=true) eikä lue tapahtumasyötettä, joten se saa poistot alla kuvatusta viikoittaisesta täydestä latauksesta.

Taulut

Yksinkertainen rakenne, jossa on yksi taulu kullekin objektille ja yksi taulu tapahtumille:

Tallenna jokainen objekti sellaisenaan JSON-sarakkeeseen ja lisäksi erillisiin sarakkeisiin ne kentät, joilla kyselet usein (updated_at, collection_id, status, owner_id). API:n myöhemmissä versioissa ilmestyy uusia kenttiä; JSON-sarakkeen ansiosta saat ne muuttamatta työtä.

Ensimmäinen lataus

Lue jokainen lista kokonaan ilman updated_since-parametria ja kirjoita jokainen objekti. Tallenna hetki, jolloin lataus alkoi: seuraava ajo jatkaa siitä.

Inkrementaaliset lataukset

Jokaisella ajokerralla, esimerkiksi 15 minuutin välein:

1. Tallenna tämän ajon alkamisaika.
2. Lue kortit niin, että updated_since on edellisen ajon alkamisaika miinus minuutin varmuusmarginaali, ja pyydä jokainen seuraava sivu kursorilla ja samalla updated_since-parametrilla (kursori ilman ensimmäisen sivun suodattimia palauttaa 400 api.cursor_invalid). Lue kokoelmat, käyttäjät ja ryhmät kokonaan: kokoelmilla ja käyttäjillä ei ole updated_at-kenttää, eikä ryhmän updated_at muutu, kun sen jäsenet muuttuvat. Luo tai korvaa jokainen objekti id-tunnuksen mukaan.
3. Lue tapahtumasyötettä edellisellä kerralla tallentamastasi kursorista. Lisää jokainen tapahtuma sursa_events-tauluun ja toteuta poistot: card.deleted-tapahtuma poistaa kortin sursa_cards-taulusta tai merkitsee sen poistetuksi, jos säilytät historiaa.
4. Tallenna alkamisaika ja syötteen uusi kursori samassa transaktiossa tietojen kanssa, jos tietovarastosi tukee sitä.

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

Syötteen kursori palautetaan jokaisella sivulla, myös viimeisellä. Kun tallennat sen, seuraava ajo alkaa täsmälleen viimeisen tallentamasi tapahtuman jälkeen, joten mitään tapahtumaa ei lueta kahdesti eikä ohiteta.

Kopion pitäminen luotettavana

• Yli 30 päivää ilman ajoa. Syöte säilyttää tapahtumia 30 päivää. Jos työtä ei ole ajettu tätä pidempään, aja täysi lataus uudelleen ja poista tauluistasi jokainen objekti, jota täysi lataus ei palauttanut.
• Viikoittainen täysi lataus. Vaikka syöte toimisi moitteetta, kerran viikossa ajettava täysi lataus, jota verrataan tauluihisi, on edullinen tapa tarkistaa, että kopio on täydellinen.
• Mitä avain ei näe, sitä ei viedä. Työ näkee täsmälleen sen, minkä sen avain näkee. Jos kortti poistuu avaimen käyttöoikeuksien piiristä esimerkiksi siksi, että se siirrettiin rajattuun kokoelmaan, täysi lataus ei palauta sitä; käsittele se poistettuna.
• Kysymykset. Kysymyslokilla on oma vientinsä: GET /v1/questions/export.csv palauttaa ajanjakson kysymykset yhtenä CSV-tiedostona arvioineen ja lähteineen. Se edellyttää käyttöaluetta ask.

Rajojen kunnioittaminen

Suuren organisaation täysi lataus tekee paljon pyyntöjä. Lue yksi lista kerrallaan parametrilla limit=200 ja seuraa RateLimit-otsakkeita. Työ, joka ajetaan 15 minuutin välein ja lukee vain muutokset, pysyy selvästi kaikkien kiintiöiden alapuolella.