Izvoz u skladište podataka

Analitičari često žele podatke iz Surse imati uz ostale podatke tvrtke: koje su kartice provjerene i tko ih je provjerio, kako znanje raste po timovima, koja pitanja ostaju bez odgovora. Ovaj vodič gradi zakazani posao koji kopira Sursu u vaše skladište i održava kopiju ažurnom.

Posao kombinira tri dijela API-ja:

Potreban vam je ključ s opsezima cards:read i users:read te webhooks:manage za feed događaja. Poslu dodijelite vlastiti ključ s ulogom admin: puni popis korisnika i feed događaja dostupni su samo njemu. Ključ s užom ulogom i podatkovnim profilom izvozi samo ono što smije vidjeti određena skupina; on čita kratki popis korisnika (brief=true) i nema feed događaja, pa brisanja preuzima iz tjednog potpunog punjenja opisanog u nastavku.

Tablice

Jednostavan raspored s jednom tablicom po objektu i jednom tablicom za događaje:

Svaki objekt pohranite onakav kakav stiže, kao JSON stupac, uz stupce koje često upitujete (updated_at, collection_id, status, owner_id). Nova polja pojavljuju se u kasnijim verzijama API-ja; uz JSON stupac dobivate ih bez mijenjanja posla.

Prvo punjenje

Pročitajte svaki popis u cijelosti, bez updated_since, i zapišite svaki objekt. Zapamtite trenutak u kojem je punjenje započelo: sljedeće pokretanje nastavlja odande.

Inkrementalna punjenja

Pri svakom pokretanju, na primjer svakih 15 minuta:

1. Zapamtite vrijeme početka ovog pokretanja.
2. Kartice čitajte s updated_since postavljenim na vrijeme početka prethodnog pokretanja, minus jedna minuta rezerve, a svaku sljedeću stranicu tražite s kursorom i istim updated_since (kursor bez filtara prve stranice vraća 400 api.cursor_invalid). Zbirke, korisnike i grupe čitajte cijele: zbirke i korisnici nemaju updated_at, a updated_at grupe ne mijenja se kad se promijene njezini članovi. Svaki objekt stvorite ili zamijenite prema id.
3. Pročitajte feed događaja od kursora koji ste spremili prošli put. Svaki događaj dodajte u sursa_events i primijenite brisanja: događaj card.deleted uklanja karticu iz sursa_cards ili je označava kao izbrisanu ako čuvate povijest.
4. Spremite vrijeme početka i novi kursor feeda, u istoj transakciji s podacima ako vaše skladište to podržava.

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 feeda vraća se na svakoj stranici, pa i na posljednjoj. Ako ga spremite, sljedeće pokretanje počinje točno nakon posljednjeg događaja koji ste pohranili, pa se nijedan događaj ne čita dvaput niti se preskače.

Kako kopija ostaje vjerna

• Više od 30 dana bez pokretanja. Feed čuva događaje 30 dana. Ako se posao nije pokretao dulje od toga, ponovno pokrenite potpuno punjenje i iz svojih tablica uklonite svaki objekt koji potpuno punjenje nije vratilo.
• Tjedno potpuno punjenje. Čak i kad feed radi ispravno, potpuno punjenje jednom tjedno, uspoređeno s vašim tablicama, jeftina je provjera da je kopija potpuna.
• Ono što ključ ne vidi ne izvozi se. Posao vidi točno ono što vidi njegov ključ. Ako kartica izađe iz pristupa ključa, na primjer zato što je premještena u ograničenu zbirku, potpuno je punjenje neće vratiti; tretirajte je kao uklonjenu.
• Pitanja. Zapisnik pitanja ima vlastiti izvoz: GET /v1/questions/export.csv vraća pitanja iz nekog razdoblja kao jednu CSV datoteku, s ocjenama i izvorima. Potreban je opseg ask.

Poštujte ograničenja

Potpuno punjenje velike organizacije šalje mnogo zahtjeva. Čitajte jedan po jedan popis s limit=200 i pratite zaglavlja RateLimit. Posao koji se pokreće svakih 15 minuta i čita samo promjene ostaje daleko ispod svake kvote.