Export to a data warehouse

Analysts often want Sursa data next to the rest of the company's data: which cards are verified and by whom, how knowledge grows per team, which questions remain unanswered. This guide builds a scheduled job that copies Sursa into your warehouse and keeps the copy current.

The job combines three parts of the API:

You need a key with the cards:read and users:read scopes, plus webhooks:manage for the event feed. Give the job its own key with the admin role: the full list of users and the event feed are open only to it. A key with a narrower role and data profile exports only what a particular group may see; it reads the brief list of users (brief=true) and no event feed, so it takes deletions from the weekly full load below.

Tables

A simple layout with one table per object and one table for events:

Store each object as it comes, as a JSON column, plus the columns you query often (updated_at, collection_id, status, owner_id). New fields appear in later versions of the API; with the JSON column you get them without changing the job.

First load

Read every list completely, without updated_since, and write every object. Remember the moment the load started: the next run continues from there.

Incremental loads

On every run, for example every 15 minutes:

1. Remember the start time of this run.
2. Read the cards with updated_since set to the start time of the previous run, minus a minute of margin, and request each next page with the cursor and the same updated_since (a cursor without the filters of the first page returns 400 api.cursor_invalid). Read the collections, users and groups whole: collections and users have no updated_at, and the updated_at of a group does not change when its members do. Upsert each object by id.
3. Read the event feed from the cursor you saved last time. Append each event to sursa_events and apply the deletions: a card.deleted event removes the card from sursa_cards, or marks it deleted if you keep history.
4. Save the start time and the new feed cursor, in the same transaction as the data if your warehouse supports it.

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

The feed cursor is returned on every page, even the last one. Saving it means the next run starts exactly after the last event you stored, so no event is read twice or skipped.

Keep the copy honest

• More than 30 days without a run. The feed keeps events for 30 days. If the job has not run for longer, run a full load again, and remove from your tables every object that the full load did not return.
• Weekly full load. Even with a healthy feed, a full load once a week, compared with your tables, is a cheap check that the copy is complete.
• What the key cannot see is not exported. The job sees exactly what its key sees. If a card leaves the key's access, for example because it moved to a restricted collection, a full load will not return it; handle it as removed.
• Questions. The question log has its own export: GET /v1/questions/export.csv returns the questions of a period as one CSV file, with ratings and sources. It needs the ask scope.

Respect the limits

A full load of a large organization makes many requests. Read one list at a time with limit=200, and watch the RateLimit headers. A job every 15 minutes that only reads changes stays far below any quota.