Exportálás adattárházba

Az elemzők gyakran a cég többi adata mellett szeretnék látni a Sursa adatait is: mely kártyák ellenőrzöttek és ki ellenőrizte őket, hogyan gyarapodik a tudás csapatonként, mely kérdések maradnak megválaszolatlanul. Ez az útmutató egy ütemezett feladatot épít, amely a Sursa adatait az adattárházadba másolja, és naprakészen tartja a másolatot.

A feladat az API három részét kombinálja:

Olyan kulcsra van szükséged, amely cards:read és users:read hatókörrel, valamint az eseménycsatornához webhooks:manage hatókörrel rendelkezik. Adj a feladatnak saját, admin szerepkörű kulcsot: a felhasználók teljes listája és az eseménycsatorna csak neki érhető el. A szűkebb szerepkörű és adatprofilú kulcs csak azt exportálja, amit egy bizonyos csoport láthat; a felhasználók rövid listáját olvassa (brief=true), eseménycsatornát nem, ezért a törléseket a lent leírt heti teljes betöltésből veszi át.

Táblák

Egyszerű elrendezés objektumonként egy táblával és egy táblával az eseményeknek:

Minden objektumot úgy tárolj, ahogy megérkezik, JSON-oszlopban, továbbá a gyakran lekérdezett oszlopokban (updated_at, collection_id, status, owner_id). Az API későbbi verzióiban új mezők jelennek meg; a JSON-oszloppal ezeket a feladat módosítása nélkül is megkapod.

Első betöltés

Olvass be minden listát teljesen, updated_since nélkül, és írj ki minden objektumot. Jegyezd meg, mikor indult a betöltés: a következő futás innen folytatja.

Növekményes betöltések

Minden futáskor, például 15 percenként:

1. Jegyezd meg ennek a futásnak a kezdési idejét.
2. A kártyákat olvasd be úgy, hogy az updated_since az előző futás kezdési ideje legyen, egy perc ráhagyással, és minden következő oldalt a kurzorral és ugyanazzal az updated_since értékkel kérj le (az első oldal szűrői nélküli kurzorra a válasz 400 api.cursor_invalid). A gyűjteményeket, a felhasználókat és a csoportokat teljes egészükben olvasd be: a gyűjteményeknek és a felhasználóknak nincs updated_at mezőjük, a csoport updated_at mezője pedig nem változik, amikor a tagjai változnak. Minden objektumot az id szerint hozz létre vagy cserélj le.
3. Olvasd be az eseménycsatornát a legutóbb elmentett kurzortól. Minden eseményt fűzz hozzá a sursa_events táblához, és vezesd át a törléseket: a card.deleted esemény eltávolítja a kártyát a sursa_cards táblából, vagy ha megőrzöd az előzményeket, töröltként jelöli meg.
4. Mentsd el a kezdési időt és a csatorna új kurzorát, ha az adattárházad támogatja, ugyanabban a tranzakcióban, mint az adatokat.

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

A csatorna kurzora minden oldalon megérkezik, az utolsón is. Ha elmented, a következő futás pontosan az utolsó eltárolt esemény után kezd, így egyetlen eseményt sem olvasol be kétszer, és egyet sem hagysz ki.

A másolat pontosságának megőrzése

• Több mint 30 nap futás nélkül. A csatorna 30 napig őrzi az eseményeket. Ha a feladat ennél régebben nem futott, futtass újra teljes betöltést, és távolíts el a tábláidból minden olyan objektumot, amelyet a teljes betöltés nem adott vissza.
• Heti teljes betöltés. A csatorna hibátlan működése mellett is olcsó ellenőrzés, hogy a másolat teljes-e, ha hetente egyszer teljes betöltést futtatsz, és összeveted a tábláiddal.
• Amit a kulcs nem lát, az nem kerül exportálásra. A feladat pontosan azt látja, amit a kulcsa. Ha egy kártya kikerül a kulcs hozzáféréséből, például mert egy korlátozott gyűjteménybe került, a teljes betöltés nem adja vissza; kezeld eltávolítottként.
• Kérdések. A kérdésnaplónak saját exportja van: a GET /v1/questions/export.csv egy időszak kérdéseit egyetlen CSV-fájlban adja vissza, az értékelésekkel és a forrásokkal együtt. Ehhez az ask hatókör szükséges.

A korlátok betartása

Egy nagy szervezet teljes betöltése sok kérést igényel. Egyszerre egy listát olvass be limit=200 értékkel, és figyeld a RateLimit fejléceket. Egy 15 percenként futó, csak a változásokat olvasó feladat messze minden kvóta alatt marad.