Exportar a un almacén de datos

Los analistas suelen querer los datos de Sursa junto al resto de los datos de la empresa: qué tarjetas están verificadas y por quién, cómo crece el conocimiento por equipo, qué preguntas siguen sin respuesta. Esta guía crea un trabajo programado que copia Sursa en su almacén de datos y mantiene la copia al día.

El trabajo combina tres partes de la API:

Necesita una clave con los ámbitos cards:read y users:read, además de webhooks:manage para el feed de eventos. Dé al trabajo su propia clave con el rol admin: la lista completa de usuarios y el feed de eventos solo están abiertos para ella. Una clave con un rol y un perfil de datos más restringidos exporta solo lo que puede ver un grupo concreto; lee la lista breve de usuarios (brief=true) y no lee el feed de eventos, así que obtiene las eliminaciones de la carga completa semanal que se describe más abajo.

Tablas

Una estructura sencilla con una tabla por objeto y una tabla para los eventos:

Guarde cada objeto tal como llega, en una columna JSON, además de las columnas que consulta a menudo (updated_at, collection_id, status, owner_id). En versiones posteriores de la API aparecen campos nuevos; con la columna JSON los recibe sin cambiar el trabajo.

Carga inicial

Lea cada lista completa, sin updated_since, y escriba todos los objetos. Recuerde el momento en que empezó la carga: la ejecución siguiente continúa desde ahí.

Cargas incrementales

En cada ejecución, por ejemplo cada 15 minutos:

1. Recuerde la hora de inicio de esta ejecución.
2. Lea las tarjetas con updated_since igual a la hora de inicio de la ejecución anterior, menos un minuto de margen, y pida cada página siguiente con el cursor y el mismo updated_since (un cursor sin los filtros de la primera página devuelve 400 api.cursor_invalid). Lea completas las colecciones, los usuarios y los grupos: las colecciones y los usuarios no tienen updated_at, y el updated_at de un grupo no cambia cuando cambian sus miembros. Inserte o actualice cada objeto por su id.
3. Lea el feed de eventos desde el cursor que guardó la última vez. Añada cada evento a sursa_events y aplique las eliminaciones: un evento card.deleted elimina la tarjeta de sursa_cards o la marca como eliminada si conserva el historial.
4. Guarde la hora de inicio y el nuevo cursor del feed en la misma transacción que los datos, si su almacén lo admite.

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 feed se devuelve en todas las páginas, incluso en la última. Si lo guarda, la ejecución siguiente empieza justo después del último evento que almacenó, así que ningún evento se lee dos veces ni se omite.

Mantener la copia fiel

• Más de 30 días sin ejecutarse. El feed conserva los eventos durante 30 días. Si el trabajo lleva más tiempo sin ejecutarse, vuelva a hacer una carga completa y elimine de sus tablas todos los objetos que la carga completa no devolvió.
• Carga completa semanal. Incluso con un feed en buen estado, una carga completa una vez a la semana, comparada con sus tablas, es una comprobación barata de que la copia está completa.
• Lo que la clave no puede ver no se exporta. El trabajo ve exactamente lo que ve su clave. Si una tarjeta sale del acceso de la clave, por ejemplo porque se trasladó a una colección restringida, una carga completa no la devolverá; trátela como eliminada.
• Preguntas. El registro de preguntas tiene su propia exportación: GET /v1/questions/export.csv devuelve las preguntas de un período en un solo archivo CSV, con las valoraciones y las fuentes. Requiere el ámbito ask.

Respetar los límites

Una carga completa de una organización grande hace muchas solicitudes. Lea las listas de una en una con limit=200 y vigile los encabezados RateLimit. Un trabajo cada 15 minutos que solo lee los cambios queda muy por debajo de cualquier cuota.