Exportar a un magatzem de dades

Sovint, els equips d’anàlisi volen tenir les dades de Sursa al costat de la resta de dades de l’empresa: quines entrades estan verificades i qui les ha verificat, com creix el coneixement per equip, quines preguntes continuen sense resposta. Aquesta guia construeix un treball programat que copia Sursa al vostre magatzem de dades i en manté la còpia actualitzada.

El treball combina tres parts de l’API:

Necessiteu una clau amb els àmbits cards:read i users:read, i webhooks:manage per al canal d’esdeveniments. Doneu al treball la seva pròpia clau amb el rol admin: la llista completa d’usuaris i el canal d’esdeveniments només són accessibles amb aquest rol. Una clau amb un rol i un perfil de dades més restringits només exporta el que pot veure un grup concret; llegeix la llista breu d’usuaris (brief=true) i cap canal d’esdeveniments, de manera que obté les eliminacions de la càrrega completa setmanal que es descriu més avall.

Taules

Una estructura senzilla, amb una taula per objecte i una taula per als esdeveniments:

Deseu cada objecte tal com arriba, en una columna JSON, més les columnes que consulteu sovint (updated_at, collection_id, status, owner_id). En versions posteriors de l’API apareixen camps nous; amb la columna JSON, els obteniu sense canviar el treball.

Càrrega inicial

Llegiu totes les llistes senceres, sense updated_since, i escriviu tots els objectes. Recordeu el moment en què ha començat la càrrega: l’execució següent continua a partir d’aquí.

Càrregues incrementals

A cada execució, per exemple cada 15 minuts:

1. Recordeu l’hora d’inici d’aquesta execució.
2. Llegiu les entrades amb updated_since igual a l’hora d’inici de l’execució anterior, menys un minut de marge, i demaneu cada pàgina següent amb el cursor i el mateix updated_since (un cursor sense els filtres de la primera pàgina retorna 400 api.cursor_invalid). Llegiu senceres les col·leccions, els usuaris i els grups: les col·leccions i els usuaris no tenen updated_at, i el updated_at d’un grup no canvia quan en canvien els membres. Inseriu o actualitzeu cada objecte per id.
3. Llegiu el canal d’esdeveniments a partir del cursor que vau desar l’última vegada. Afegiu cada esdeveniment a sursa_events i apliqueu les eliminacions: un esdeveniment card.deleted elimina l’entrada de sursa_cards, o la marca com a eliminada si conserveu l’historial.
4. Deseu l’hora d’inici i el nou cursor del canal, en la mateixa transacció que les dades si el vostre magatzem ho admet.

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

El cursor del canal es retorna a cada pàgina, fins i tot a l’última. Si el deseu, l’execució següent comença exactament després de l’últim esdeveniment que heu desat, de manera que cap esdeveniment no es llegeix dues vegades ni se salta.

Mantenir la còpia fidel

• Més de 30 dies sense cap execució. El canal conserva els esdeveniments durant 30 dies. Si el treball no s’ha executat durant més temps, torneu a fer una càrrega completa i elimineu de les vostres taules tots els objectes que la càrrega completa no hagi retornat.
• Càrrega completa setmanal. Fins i tot amb un canal en bon estat, una càrrega completa un cop per setmana, comparada amb les vostres taules, és una manera barata de comprovar que la còpia és completa.
• El que la clau no pot veure no s’exporta. El treball veu exactament el que veu la seva clau. Si una entrada surt de l’accés de la clau, per exemple perquè s’ha mogut a una col·lecció restringida, una càrrega completa no la retornarà; tracteu-la com a eliminada.
• Preguntes. El registre de preguntes té la seva pròpia exportació: GET /v1/questions/export.csv retorna les preguntes d’un període en un sol fitxer CSV, amb les valoracions i les fonts. Necessita l’àmbit ask.

Respectar els límits

Una càrrega completa d’una organització gran fa moltes sol·licituds. Llegiu les llistes d’una en una amb limit=200 i vigileu les capçaleres RateLimit. Un treball cada 15 minuts que només llegeix els canvis queda molt per sota de qualsevol quota.