Exportar para um data warehouse

Os analistas muitas vezes querem os dados do Sursa ao lado dos outros dados da empresa: quais cartões estão verificados e por quem, como o conhecimento cresce em cada equipe, quais perguntas continuam sem resposta. Este guia monta um job agendado que copia o Sursa para o seu data warehouse e mantém a cópia atualizada.

O job combina três partes da API:

Você precisa de uma chave com os escopos cards:read e users:read, além de webhooks:manage para o feed de eventos. Dê ao job uma chave própria com a função admin: só ela tem acesso à lista completa de usuários e ao feed de eventos. Uma chave com uma função e um perfil de dados mais restritos exporta só o que um determinado grupo pode ver; ela lê a lista resumida de usuários (brief=true) e não lê o feed de eventos, então obtém as exclusões da carga completa semanal descrita abaixo.

Tabelas

Uma estrutura simples, com uma tabela por objeto e uma tabela para os eventos:

Guarde cada objeto como ele chega, em uma coluna JSON, mais as colunas que você consulta com frequência (updated_at, collection_id, status, owner_id). Novos campos aparecem em versões posteriores da API; com a coluna JSON, você os recebe sem alterar o job.

Carga inicial

Leia cada lista por inteiro, sem updated_since, e grave todos os objetos. Guarde o momento em que a carga começou: a próxima execução continua a partir dele.

Cargas incrementais

A cada execução, por exemplo a cada 15 minutos:

1. Guarde o horário de início desta execução.
2. Leia os cartões com updated_since igual ao horário de início da execução anterior, menos um minuto de margem, e peça cada página seguinte com o cursor e o mesmo updated_since (um cursor sem os filtros da primeira página retorna 400 api.cursor_invalid). Leia as coleções, os usuários e os grupos por inteiro: as coleções e os usuários não têm updated_at, e o updated_at de um grupo não muda quando os membros dele mudam. Insira ou atualize cada objeto pelo id.
3. Leia o feed de eventos a partir do cursor que você salvou da última vez. Acrescente cada evento a sursa_events e aplique as exclusões: um evento card.deleted remove o cartão de sursa_cards ou o marca como excluído, se você mantém o histórico.
4. Salve o horário de início e o novo cursor do feed na mesma transação que os dados, se o seu data warehouse permitir.

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

O cursor do feed é retornado em todas as páginas, inclusive na última. Salvá-lo faz a próxima execução começar exatamente depois do último evento que você guardou, de modo que nenhum evento é lido duas vezes nem pulado.

Manter a cópia fiel

• Mais de 30 dias sem execução. O feed mantém os eventos por 30 dias. Se o job ficou mais tempo sem rodar, execute uma carga completa de novo e remova das suas tabelas todo objeto que a carga completa não retornou.
• Carga completa semanal. Mesmo com o feed em ordem, uma carga completa uma vez por semana, comparada com as suas tabelas, é uma forma barata de verificar se a cópia está completa.
• O que a chave não pode ver não é exportado. O job vê exatamente o que a chave dele vê. Se um cartão sai do acesso da chave, por exemplo porque foi movido para uma coleção restrita, uma carga completa não vai retorná-lo; trate-o como removido.
• Perguntas. O registro de perguntas tem uma exportação própria: GET /v1/questions/export.csv retorna as perguntas de um período em um único arquivo CSV, com avaliações e fontes. Ela exige o escopo ask.

Respeitar os limites

Uma carga completa de uma organização grande faz muitas requisições. Leia uma lista por vez com limit=200 e acompanhe os cabeçalhos RateLimit. Um job a cada 15 minutos que só lê as alterações fica muito abaixo de qualquer cota.