डेटा वेयरहाउस में एक्सपोर्ट

एनालिस्ट अक्सर 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 मिनट में चलने वाला जॉब, जो सिर्फ़ बदलाव पढ़ता है, किसी भी कोटा से बहुत नीचे रहता है।