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

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