Export till ett datalager

Analytiker vill ofta ha Sursas data bredvid företagets övriga data: vilka kort som är verifierade och av vem, hur kunskapen växer per team, vilka frågor som förblir obesvarade. Den här guiden bygger ett schemalagt jobb som kopierar Sursa till ditt datalager och håller kopian aktuell.

Jobbet kombinerar tre delar av API:et:

Du behöver en nyckel med omfången cards:read och users:read samt webhooks:manage för händelseflödet. Ge jobbet en egen nyckel med rollen admin: den fullständiga listan över användare och händelseflödet är bara öppna för den. En nyckel med en snävare roll och dataprofil exporterar bara det som en viss grupp får se; den läser den korta listan över användare (brief=true) och inget händelseflöde, så den får borttagningarna från den fullständiga inläsningen varje vecka som beskrivs nedan.

Tabeller

En enkel struktur med en tabell per objekt och en tabell för händelser:

Lagra varje objekt som det kommer, i en JSON-kolumn, plus de kolumner som du ofta frågar på (updated_at, collection_id, status, owner_id). Nya fält tillkommer i senare versioner av API:et; med JSON-kolumnen får du dem utan att ändra jobbet.

Första inläsningen

Läs varje lista i sin helhet, utan updated_since, och skriv varje objekt. Spara tidpunkten då inläsningen startade: nästa körning fortsätter därifrån.

Inkrementella inläsningar

Vid varje körning, till exempel var 15:e minut:

1. Spara starttiden för den här körningen.
2. Läs korten med updated_since satt till starttiden för den föregående körningen, minus en minuts marginal, och hämta varje nästa sida med markören och samma updated_since (en markör utan filtren från den första sidan returnerar 400 api.cursor_invalid). Läs samlingarna, användarna och grupperna i sin helhet: samlingar och användare har inget updated_at, och en grupps updated_at ändras inte när dess medlemmar ändras. Skapa eller ersätt (upsert) varje objekt efter id.
3. Läs händelseflödet från markören som du sparade förra gången. Lägg till varje händelse i sursa_events och utför borttagningarna: en händelse card.deleted tar bort kortet från sursa_cards, eller markerar det som borttaget om du sparar historik.
4. Spara starttiden och flödets nya markör, i samma transaktion som data om ditt datalager stöder det.

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

Flödets markör returneras på varje sida, även den sista. Sparar du den startar nästa körning exakt efter den senaste händelsen som du lagrade, så att ingen händelse läses två gånger eller hoppas över.

Håll kopian korrekt

• Mer än 30 dagar utan körning. Flödet sparar händelser i 30 dagar. Om jobbet inte har körts på längre tid kör du en fullständig inläsning igen och tar bort varje objekt ur dina tabeller som den fullständiga inläsningen inte returnerade.
• Fullständig inläsning varje vecka. Även med ett fungerande flöde är en fullständig inläsning en gång i veckan, jämförd med dina tabeller, en billig kontroll av att kopian är fullständig.
• Det som nyckeln inte kan se exporteras inte. Jobbet ser exakt det som dess nyckel ser. Om ett kort hamnar utanför nyckelns åtkomst, till exempel för att det har flyttats till en begränsad samling, returneras det inte av en fullständig inläsning; behandla det som borttaget.
• Frågor. Frågeloggen har en egen export: GET /v1/questions/export.csv returnerar frågorna för en period som en enda CSV-fil, med betyg och källor. Den kräver omfånget ask.

Respektera gränserna

En fullständig inläsning av en stor organisation gör många anrop. Läs en lista i taget med limit=200 och håll koll på RateLimit-huvudena. Ett jobb var 15:e minut som bara läser ändringar ligger långt under alla kvoter.