التصدير إلى مستودع البيانات

كثيرًا ما يريد المحللون بيانات 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 دقيقة ولا تقرأ إلا التغييرات تبقى بعيدة جدًا عن أي حصة.