Eksportuj dane do hurtowni danych

Analitycy często chcą mieć dane Sursy obok pozostałych danych firmy: które karty są zweryfikowane i przez kogo, jak przybywa wiedzy w poszczególnych zespołach, które pytania pozostają bez odpowiedzi. Ten przewodnik buduje zadanie uruchamiane według harmonogramu, które kopiuje dane Sursy do Twojej hurtowni i utrzymuje kopię w aktualnym stanie.

Zadanie łączy trzy części API:

Potrzebujesz klucza z zakresami cards:read i users:read oraz webhooks:manage do kanału zdarzeń. Daj zadaniu własny klucz z rolą admin: tylko dla niego dostępne są pełna lista użytkowników i kanał zdarzeń. Klucz z węższą rolą i profilem dostępu eksportuje tylko to, co może widzieć określona grupa; odczytuje skróconą listę użytkowników (brief=true) i nie ma dostępu do kanału zdarzeń, dlatego usunięcia przejmuje z cotygodniowego pełnego ładowania opisanego niżej.

Tabele

Prosty układ z jedną tabelą na każdy obiekt i jedną tabelą na zdarzenia:

Zapisuj każdy obiekt w otrzymanej postaci jako kolumnę JSON, a obok niej kolumny, po których często filtrujesz (updated_at, collection_id, status, owner_id). Nowe pola pojawiają się w późniejszych wersjach API; dzięki kolumnie JSON otrzymujesz je bez zmiany zadania.

Pierwsze ładowanie

Odczytaj w całości każdą listę, bez updated_since, i zapisz każdy obiekt. Zapamiętaj moment, w którym ładowanie się rozpoczęło: od tego momentu będzie kontynuować następne uruchomienie.

Ładowania przyrostowe

Przy każdym uruchomieniu, np. co 15 minut:

1. Zapamiętaj czas rozpoczęcia tego uruchomienia.
2. Odczytaj karty z updated_since ustawionym na czas rozpoczęcia poprzedniego uruchomienia minus minuta zapasu i każdą następną stronę pobieraj z kursorem i tym samym updated_since (kursor bez filtrów pierwszej strony zwraca 400 api.cursor_invalid). Kolekcje, użytkowników i grupy odczytuj w całości: kolekcje i użytkownicy nie mają updated_at, a updated_at grupy nie zmienia się, gdy zmieniają się jej członkowie. Każdy obiekt utwórz lub zastąp według id.
3. Odczytaj kanał zdarzeń od kursora zapisanego ostatnim razem. Dopisz każde zdarzenie do sursa_events i zastosuj usunięcia: zdarzenie card.deleted usuwa kartę z sursa_cards lub oznacza ją jako usuniętą, jeśli przechowujesz historię.
4. Zapisz czas rozpoczęcia i nowy kursor kanału – w tej samej transakcji co dane, jeśli Twoja hurtownia to obsługuje.

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

Kursor kanału jest zwracany na każdej stronie, także na ostatniej. Dzięki jego zapisaniu następne uruchomienie zaczyna się dokładnie po ostatnim zapisanym zdarzeniu, więc żadne zdarzenie nie zostanie odczytane dwukrotnie ani pominięte.

Dbaj o wierność kopii

• Ponad 30 dni bez uruchomienia. Kanał przechowuje zdarzenia przez 30 dni. Jeśli zadanie nie było uruchamiane dłużej, wykonaj ponownie pełne ładowanie i usuń ze swoich tabel każdy obiekt, którego pełne ładowanie nie zwróciło.
• Cotygodniowe pełne ładowanie. Nawet przy sprawnie działającym kanale pełne ładowanie raz w tygodniu, porównane z Twoimi tabelami, to tani sposób sprawdzenia, czy kopia jest kompletna.
• To, czego klucz nie widzi, nie jest eksportowane. Zadanie widzi dokładnie to, co widzi jego klucz. Jeśli karta wypadnie poza dostęp klucza, np. dlatego, że przeniesiono ją do kolekcji z ograniczonym dostępem, pełne ładowanie jej nie zwróci; traktuj ją jako usuniętą.
• Pytania. Dziennik pytań ma własny eksport: GET /v1/questions/export.csv zwraca pytania z danego okresu jako jeden plik CSV, z ocenami i źródłami. Wymaga zakresu ask.

Przestrzegaj limitów

Pełne ładowanie dużej organizacji wysyła wiele żądań. Odczytuj jedną listę naraz z limit=200 i obserwuj nagłówki RateLimit. Zadanie uruchamiane co 15 minut, które odczytuje tylko zmiany, pozostaje daleko poniżej każdego przydziału.