Export do dátového skladu

Analytici často chcú mať údaje Sursy vedľa ostatných údajov firmy: ktoré karty sú overené a kým, ako rastú znalosti v jednotlivých tímoch, ktoré otázky zostávajú bez odpovede. Tento návod vytvorí plánovanú úlohu, ktorá kopíruje údaje zo Sursy do vášho dátového skladu a udržiava kópiu aktuálnu.

Úloha kombinuje tri časti API:

Potrebujete kľúč s rozsahmi cards:read a users:read a navyše webhooks:manage pre kanál udalostí. Dajte úlohe vlastný kľúč s rolou admin: úplný zoznam používateľov a kanál udalostí sú prístupné len jemu. Kľúč s užšou rolou a dátovým profilom exportuje len to, čo smie vidieť určitá skupina; číta skrátený zoznam používateľov (brief=true) a kanál udalostí nečíta, preto vymazania zisťuje z týždenného úplného načítania opísaného nižšie.

Tabuľky

Jednoduché rozloženie s jednou tabuľkou na každý objekt a jednou tabuľkou na udalosti:

Každý objekt uložte tak, ako prišiel, do stĺpca JSON a k tomu stĺpce, podľa ktorých často dopytujete (updated_at, collection_id, status, owner_id). V neskorších verziách API pribúdajú nové polia; so stĺpcom JSON ich dostanete bez zmeny úlohy.

Prvé načítanie

Prečítajte každý zoznam celý, bez updated_since, a zapíšte každý objekt. Zapamätajte si okamih, keď sa načítanie začalo: ďalší beh pokračuje odtiaľ.

Prírastkové načítania

Pri každom behu, napríklad každých 15 minút:

1. Zapamätajte si čas začiatku tohto behu.
2. Karty čítajte s updated_since nastaveným na čas začiatku predchádzajúceho behu mínus minúta rezervy a každú ďalšiu stránku si vyžiadajte s kurzorom a s tým istým updated_since (kurzor bez filtrov prvej stránky vráti 400 api.cursor_invalid). Kolekcie, používateľov a skupiny čítajte celé: kolekcie a používatelia nemajú updated_at a updated_at skupiny sa pri zmene jej členov nemení. Každý objekt vytvorte alebo nahraďte podľa id.
3. Prečítajte kanál udalostí od kurzora, ktorý ste si uložili naposledy. Každú udalosť pripíšte do sursa_events a premietnite vymazania: udalosť card.deleted odstráni kartu zo sursa_cards alebo ju označí ako vymazanú, ak uchovávate históriu.
4. Uložte čas začiatku a nový kurzor kanála, a to v tej istej transakcii ako údaje, ak to váš sklad podporuje.

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

Kurzor kanála sa vracia na každej stránke, aj na poslednej. Keď ho uložíte, ďalší beh začne presne za poslednou udalosťou, ktorú ste uložili, takže žiadnu udalosť neprečítate dvakrát ani ju nepreskočíte.

Aby kópia zodpovedala skutočnosti

• Viac ako 30 dní bez behu. Kanál uchováva udalosti 30 dní. Ak úloha nebežala dlhšie, spustite znova úplné načítanie a zo svojich tabuliek odstráňte každý objekt, ktorý úplné načítanie nevrátilo.
• Týždenné úplné načítanie. Aj keď kanál funguje bez problémov, úplné načítanie raz týždenne porovnané s vašimi tabuľkami je lacná kontrola, že kópia je úplná.
• Čo kľúč nevidí, sa neexportuje. Úloha vidí presne to, čo vidí jej kľúč. Ak sa karta dostane mimo prístupu kľúča, napríklad preto, že sa presunula do kolekcie s obmedzeným prístupom, úplné načítanie ju nevráti; považujte ju za odstránenú.
• Otázky. Log otázok má vlastný export: GET /v1/questions/export.csv vráti otázky za obdobie ako jeden súbor CSV s hodnoteniami a zdrojmi. Vyžaduje rozsah ask.

Dodržiavajte limity

Úplné načítanie veľkej organizácie posiela veľa požiadaviek. Čítajte vždy len jeden zoznam s limit=200 a sledujte hlavičky RateLimit. Úloha, ktorá beží každých 15 minút a číta len zmeny, zostáva hlboko pod každou kvótou.