Извоз у складиште података

Аналитичари често желе да подаци из платформе Sursa буду поред осталих података компаније: које су картице проверене и ко их је проверио, како знање расте по тимовима, која питања остају без одговора. Овај водич гради заказани посао који копира податке из платформе Sursa у ваше складиште и одржава копију ажурном.

Посао комбинује три дела API интерфејса:

Потребан вам је кључ са опсезима cards:read и users:read, плус webhooks:manage за фид догађаја. Дајте послу сопствени кључ са улогом admin: само њему су доступни пун списак корисника и фид догађаја. Кључ са ужом улогом и профилом приступа извози само оно што одређена група сме да види; он чита скраћени списак корисника (brief=true) и нема фид догађаја, па брисања преузима из недељног потпуног пуњења описаног у наставку.

Табеле

Једноставан распоред, са по једном табелом за сваки објекат и једном табелом за догађаје:

Сваки објекат чувајте онакав какав стиже, у JSON колони, плус колоне по којима често постављате упите (updated_at, collection_id, status, owner_id). Нова поља се појављују у каснијим верзијама API интерфејса; уз JSON колону добијате их без измене посла.

Прво пуњење

Прочитајте сваку листу у целости, без updated_since, и упишите сваки објекат. Запамтите тренутак када је пуњење почело: следеће покретање наставља одатле.

Инкрементална пуњења

При сваком покретању, на пример на сваких 15 минута:

1. Запамтите време почетка овог покретања.
2. Картице читајте са updated_since постављеним на време почетка претходног покретања, минус један минут резерве, а сваку следећу страницу затражите са курсором и истим updated_since (курсор без филтера прве странице враћа 400 api.cursor_invalid). Колекције, кориснике и групе читајте целе: колекције и корисници немају updated_at, а updated_at групе се не мења када се мењају њени чланови. Сваки објекат креирајте или замените по id.
3. Прочитајте фид догађаја од курсора који сте сачували прошли пут. Сваки догађај додајте у sursa_events и примените брисања: догађај card.deleted уклања картицу из sursa_cards или је означава као обрисану ако чувате историју.
4. Сачувајте време почетка и нови курсор фида, у истој трансакцији са подацима ако ваше складиште то подржава.

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

Курсор фида се враћа на свакој страници, па и на последњој. Када га сачувате, следеће покретање почиње тачно после последњег догађаја који сте уписали, па се ниједан догађај не чита двапут нити се прескаче.

Како да копија остане тачна

• Више од 30 дана без покретања. Фид чува догађаје 30 дана. Ако посао није радио дуже од тога, поново покрените потпуно пуњење и из својих табела уклоните сваки објекат који потпуно пуњење није вратило.
• Недељно потпуно пуњење. Чак и када фид ради исправно, потпуно пуњење једном недељно, упоређено са вашим табелама, јефтина је провера да је копија потпуна.
• Оно што кључ не види не извози се. Посао види тачно оно што види његов кључ. Ако картица изађе из приступа кључа, на пример зато што је премештена у ограничену колекцију, потпуно пуњење је неће вратити; третирајте је као уклоњену.
• Питања. Дневник питања има сопствени извоз: GET /v1/questions/export.csv враћа питања за одређени период као један CSV фајл, са оценама и изворима. Потребан му је опсег ask.

Поштовање ограничења

Потпуно пуњење велике организације шаље много захтева. Читајте једну по једну листу са limit=200 и пратите заглавља RateLimit. Посао на сваких 15 минута који чита само измене остаје далеко испод било које квоте.