Exporter vers un entrepôt de données

Les analystes veulent souvent disposer des données de Sursa à côté des autres données de l’entreprise : quelles fiches sont validées et par qui, comment les connaissances progressent par équipe, quelles questions restent sans réponse. Ce guide construit un job planifié qui copie Sursa dans votre entrepôt et maintient la copie à jour.

Le job combine trois parties de l’API :

Il vous faut une clé dotée des portées cards:read et users:read, ainsi que de webhooks:manage pour le flux d’événements. Donnez au job sa propre clé avec le rôle admin : la liste complète des utilisateurs et le flux d’événements ne sont ouverts qu’à ce rôle. Une clé dotée d’un rôle et d’un profil de données plus restreints n’exporte que ce qu’un groupe particulier peut voir ; elle lit la liste abrégée des utilisateurs (brief=true) et aucun flux d’événements : elle tire donc les suppressions du chargement complet hebdomadaire décrit ci-dessous.

Tables

Une organisation simple, avec une table par objet et une table pour les événements :

Stockez chaque objet tel qu’il arrive, dans une colonne JSON, avec en plus les colonnes que vous interrogez souvent (updated_at, collection_id, status, owner_id). De nouveaux champs apparaissent dans les versions ultérieures de l’API ; grâce à la colonne JSON, vous les recevez sans modifier le job.

Chargement initial

Lisez chaque liste en entier, sans updated_since, et écrivez chaque objet. Retenez le moment où le chargement a commencé : l’exécution suivante reprend à partir de là.

Chargements incrémentaux

À chaque exécution, par exemple toutes les 15 minutes :

1. Retenez l’heure de début de cette exécution.
2. Lisez les fiches avec updated_since égal à l’heure de début de l’exécution précédente, moins une minute de marge, et demandez chaque page suivante avec le curseur et le même updated_since (un curseur sans les filtres de la première page renvoie 400 api.cursor_invalid). Lisez en entier les collections, les utilisateurs et les groupes : les collections et les utilisateurs n’ont pas de updated_at, et le updated_at d’un groupe ne change pas quand ses membres changent. Écrivez chaque objet selon son id (création ou remplacement).
3. Lisez le flux d’événements à partir du curseur enregistré la dernière fois. Ajoutez chaque événement à sursa_events et appliquez les suppressions : un événement card.deleted retire la fiche de sursa_cards, ou la marque comme supprimée si vous conservez l’historique.
4. Enregistrez l’heure de début et le nouveau curseur du flux, dans la même transaction que les données si votre entrepôt le permet.

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

Le curseur du flux est renvoyé sur chaque page, même la dernière. En l’enregistrant, vous faites démarrer l’exécution suivante exactement après le dernier événement stocké : aucun événement n’est lu deux fois ni sauté.

Garder une copie fidèle

• Plus de 30 jours sans exécution. Le flux conserve les événements pendant 30 jours. Si le job n’a pas tourné depuis plus longtemps, refaites un chargement complet et retirez de vos tables chaque objet que ce chargement n’a pas renvoyé.
• Chargement complet hebdomadaire. Même avec un flux en bon état, un chargement complet une fois par semaine, comparé à vos tables, est un moyen peu coûteux de vérifier que la copie est complète.
• Ce que la clé ne voit pas n’est pas exporté. Le job voit exactement ce que voit sa clé. Si une fiche sort du périmètre d’accès de la clé, par exemple parce qu’elle a été déplacée dans une collection restreinte, un chargement complet ne la renverra pas ; traitez-la comme supprimée.
• Questions. Le journal des questions a son propre export : GET /v1/questions/export.csv renvoie les questions d’une période dans un seul fichier CSV, avec les notes et les sources. Il exige la portée ask.

Respecter les limites

Un chargement complet d’une grande organisation fait de nombreuses requêtes. Lisez une liste à la fois avec limit=200 et surveillez les en-têtes RateLimit. Un job qui tourne toutes les 15 minutes et ne lit que les modifications reste très en dessous de tout quota.