Export do datového skladu

Analytici často chtějí mít data ze Sursy vedle ostatních dat firmy: které karty jsou ověřené a kým, jak přibývají znalosti v jednotlivých týmech, které otázky zůstávají bez odpovědi. Tento návod sestaví plánovanou úlohu, která kopíruje data ze Sursy do vašeho datového skladu a udržuje kopii aktuální.

Úloha kombinuje tři části API:

Potřebujete klíč s rozsahy cards:read a users:read a navíc webhooks:manage pro kanál událostí. Dejte úloze vlastní klíč s rolí admin: úplný seznam uživatelů a kanál událostí jsou přístupné jen jemu. Klíč s užší rolí a datovým profilem exportuje jen to, co smí vidět určitá skupina; čte zkrácený seznam uživatelů (brief=true) a žádný kanál událostí, proto smazání převezme z týdenního úplného načtení popsaného níže.

Tabulky

Jednoduché uspořádání s jednou tabulkou na každý objekt a jednou tabulkou pro události:

Každý objekt ukládejte tak, jak přijde, jako sloupec JSON, a k tomu sloupce, podle kterých se často dotazujete (updated_at, collection_id, status, owner_id). V pozdějších verzích API přibývají nová pole; díky sloupci JSON je získáte bez změny úlohy.

První načtení

Přečtěte každý seznam celý, bez updated_since, a zapište každý objekt. Zapamatujte si okamžik, kdy načítání začalo: další běh naváže odtud.

Přírůstková načtení

Při každém běhu, například každých 15 minut:

1. Zapamatujte si čas začátku tohoto běhu.
2. Karty čtěte s updated_since nastaveným na čas začátku předchozího běhu minus minuta rezervy a každou další stránku vyžádejte s kurzorem a se stejným updated_since (kurzor bez filtrů první stránky vrátí 400 api.cursor_invalid). Kolekce, uživatele a skupiny čtěte celé: kolekce a uživatelé updated_at nemají a updated_at skupiny se při změně jejích členů nemění. Každý objekt vytvořte nebo nahraďte podle id.
3. Čtěte kanál událostí od kurzoru uloženého při minulém běhu. Každou událost připojte do sursa_events a proveďte smazání: událost card.deleted odstraní kartu ze sursa_cards, nebo ji označí jako smazanou, pokud uchováváte historii.
4. Uložte čas začátku a nový kurzor kanálu, ve stejné transakci jako data, pokud 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álu se vrací na každé stránce, i na poslední. Když si ho uložíte, další běh začne přesně za poslední uloženou událostí, takže se žádná událost nepřečte dvakrát ani nepřeskočí.

Jak udržet kopii spolehlivou

• Více než 30 dní bez běhu. Kanál uchovává události 30 dní. Pokud úloha neběžela déle, spusťte znovu úplné načtení a odstraňte ze svých tabulek každý objekt, který úplné načtení nevrátilo.
• Týdenní úplné načtení. I při bezchybném kanálu je úplné načtení jednou týdně, porovnané s vašimi tabulkami, levnou kontrolou, že je kopie úplná.
• Co klíč nevidí, se neexportuje. Úloha vidí přesně to, co vidí její klíč. Pokud karta z přístupu klíče zmizí, například proto, že se přesunula do kolekce s omezeným přístupem, úplné načtení ji nevrátí; zacházejte s ní jako s odstraněnou.
• Otázky. Protokol otázek má vlastní export: GET /v1/questions/export.csv vrací otázky za zvolené období jako jeden soubor CSV, s hodnoceními a zdroji. Vyžaduje rozsah ask.

Dodržujte limity

Úplné načtení velké organizace odešle mnoho požadavků. Čtěte seznamy jeden po druhém s limit=200 a sledujte hlavičky RateLimit. Úloha spouštěná každých 15 minut, která čte jen změny, zůstává hluboko pod jakoukoli kvótou.