Eksportas į duomenų sandėlį

Analitikai dažnai nori Sursos duomenis turėti šalia kitų įmonės duomenų: kurios kortelės patikrintos ir kas jas patikrino, kaip auga kiekvienos komandos žinios, kurie klausimai lieka neatsakyti. Ši instrukcija parodo, kaip sukurti pagal tvarkaraštį vykdomą darbą, kuris kopijuoja Sursos duomenis į jūsų sandėlį ir palaiko kopiją aktualią.

Darbas sujungia tris API dalis:

Jums reikia rakto su apimtimis cards:read ir users:read, taip pat webhooks:manage įvykių juostai. Darbui suteikite atskirą raktą su vaidmeniu admin: visas naudotojų sąrašas ir įvykių juosta prieinami tik jam. Raktas su siauresniu vaidmeniu ir duomenų profiliu eksportuoja tik tai, ką gali matyti tam tikra grupė; jis skaito trumpąjį naudotojų sąrašą (brief=true) ir įvykių juostos neskaito, todėl ištrynimus gauna iš toliau aprašyto savaitinio visiško įkėlimo.

Lentelės

Paprasta struktūra – po lentelę kiekvienam objektui ir viena lentelė įvykiams:

Kiekvieną objektą saugokite tokį, koks ateina, JSON stulpelyje, ir papildomai – stulpelius, pagal kuriuos dažnai teikiate užklausas (updated_at, collection_id, status, owner_id). Vėlesnėse API versijose atsiranda naujų laukų; su JSON stulpeliu juos gausite be jokių darbo pakeitimų.

Pirmasis įkėlimas

Kiekvieną sąrašą nuskaitykite visą, be updated_since, ir įrašykite kiekvieną objektą. Įsiminkite momentą, kai įkėlimas prasidėjo: kitas paleidimas tęs nuo jo.

Inkrementiniai įkėlimai

Kiekvieno paleidimo metu, pavyzdžiui, kas 15 minučių:

1. Įsiminkite šio paleidimo pradžios laiką.
2. Korteles skaitykite su updated_since, lygiu ankstesnio paleidimo pradžios laikui, atėmus minutės atsargą, o kiekvieno kito puslapio prašykite su žymekliu ir tuo pačiu updated_since (žymeklis be pirmojo puslapio filtrų grąžina 400 api.cursor_invalid). Rinkinius, naudotojus ir grupes skaitykite visus: rinkiniai ir naudotojai neturi updated_at, o grupės updated_at nesikeičia, kai keičiasi jos nariai. Kiekvieną objektą sukurkite arba pakeiskite pagal id.
3. Skaitykite įvykių juostą nuo praėjusį kartą išsaugoto žymeklio. Kiekvieną įvykį pridėkite į sursa_events ir pritaikykite ištrynimus: įvykis card.deleted pašalina kortelę iš sursa_cards arba, jei saugote istoriją, pažymi ją kaip ištrintą.
4. Išsaugokite pradžios laiką ir naują juostos žymeklį – jei jūsų sandėlis tai palaiko, toje pačioje transakcijoje kaip ir duomenis.

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

Juostos žymeklis grąžinamas kiekviename puslapyje, net paskutiniame. Jį išsaugojus, kitas paleidimas prasideda iškart po paskutinio jūsų įrašyto įvykio, todėl joks įvykis nenuskaitomas du kartus ir nepraleidžiamas.

Kad kopija būtų patikima

• Daugiau nei 30 dienų be paleidimo. Juosta įvykius saugo 30 dienų. Jei darbas nevyko ilgiau, vėl atlikite visišką įkėlimą ir iš savo lentelių pašalinkite kiekvieną objektą, kurio visiškas įkėlimas negrąžino.
• Savaitinis visiškas įkėlimas. Net kai juosta veikia be trikdžių, kartą per savaitę atliekamas visiškas įkėlimas, palygintas su jūsų lentelėmis, – pigus būdas patikrinti, ar kopija pilna.
• Tai, ko raktas nemato, neeksportuojama. Darbas mato lygiai tai, ką mato jo raktas. Jei kortelė patenka už rakto prieigos ribų, pavyzdžiui, perkėlus ją į ribotos prieigos rinkinį, visiškas įkėlimas jos negrąžins; laikykite ją pašalinta.
• Klausimai. Klausimų žurnalas turi savo eksportą: GET /v1/questions/export.csv grąžina laikotarpio klausimus vienu CSV failu su įvertinimais ir šaltiniais. Jam reikia apimties ask.

Laikykitės limitų

Visiškas didelės organizacijos įkėlimas siunčia daug užklausų. Skaitykite po vieną sąrašą su limit=200 ir stebėkite RateLimit antraštes. Darbas kas 15 minučių, kuris skaito tik pakeitimus, lieka gerokai žemiau bet kokios kvotos.