Выгрузка в хранилище данных

Аналитикам часто нужны данные 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 минут читает только изменения, расходует лишь малую долю любой квоты.