Експорт в хранилище за данни

Анализаторите често искат данните от 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 минути, което чете само промените, остава далеч под всяка квота.