Veri ambarına aktarma

Analistler çoğu zaman Sursa verilerini şirketin diğer verilerinin yanında görmek ister: hangi kartlar doğrulanmış ve kim tarafından, bilgi ekip başına nasıl büyüyor, hangi sorular yanıtsız kalıyor. Bu kılavuz, Sursa verilerini veri ambarınıza kopyalayan ve kopyayı güncel tutan zamanlanmış bir iş oluşturur.

İş, API’nin üç bölümünü bir araya getirir:

cards:read ve users:read kapsamlarına ve ayrıca olay beslemesi için webhooks:manage kapsamına sahip bir anahtara ihtiyacınız vardır. İşe admin rolüyle kendi anahtarını verin: kullanıcıların tam listesi ve olay beslemesi yalnızca bu role açıktır. Daha dar bir rol ve veri profiline sahip bir anahtar yalnızca belirli bir grubun görebileceklerini dışa aktarır; bu anahtar kullanıcıların kısa listesini (brief=true) okur, olay beslemesini okuyamaz; bu yüzden silinenleri aşağıdaki haftalık tam yüklemeden alır.

Tablolar

Her nesne için bir tablo ve olaylar için bir tablo içeren basit bir düzen:

Her nesneyi geldiği gibi bir JSON sütununda saklayın; sık sorguladığınız alanlar için de ayrı sütunlar ekleyin (updated_at, collection_id, status, owner_id). API’nin sonraki sürümlerinde yeni alanlar ortaya çıkar; JSON sütunu sayesinde bu alanları işi değiştirmeden alırsınız.

İlk yükleme

Her listeyi updated_since olmadan baştan sona okuyun ve her nesneyi yazın. Yüklemenin başladığı anı kaydedin: sonraki çalıştırma oradan devam eder.

Artımlı yüklemeler

Her çalıştırmada, örneğin 15 dakikada bir:

1. Bu çalıştırmanın başlangıç zamanını kaydedin.
2. Kartları, updated_since değerine önceki çalıştırmanın başlangıç zamanından bir dakikalık pay düşülmüş hâlini vererek okuyun ve sonraki her sayfayı imleçle ve aynı updated_since değeriyle isteyin (ilk sayfanın filtreleri olmadan gönderilen imleç 400 api.cursor_invalid döndürür). Koleksiyonları, kullanıcıları ve grupları bütün olarak okuyun: koleksiyonların ve kullanıcıların updated_at alanı yoktur, bir grubun updated_at değeri ise üyeleri değiştiğinde değişmez. Her nesneyi id değerine göre oluşturun ya da değiştirin.
3. Olay beslemesini son kez kaydettiğiniz imleçten itibaren okuyun. Her olayı sursa_events tablosuna ekleyin ve silinenleri uygulayın: bir card.deleted olayı kartı sursa_cards tablosundan kaldırır ya da geçmişi tutuyorsanız kartı silinmiş olarak işaretler.
4. Başlangıç zamanını ve yeni besleme imlecini, veri ambarınız destekliyorsa verilerle aynı veritabanı işleminde (transaction) kaydedin.

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})

Besleme imleci, sonuncusu dahil her sayfada döndürülür. Onu kaydettiğinizde sonraki çalıştırma, sakladığınız son olaydan hemen sonra başlar; böylece hiçbir olay iki kez okunmaz ya da atlanmaz.

Kopyayı doğru tutma

• 30 günden uzun süre çalıştırma yapılmaması. Besleme olayları 30 gün saklar. İş daha uzun süre çalışmadıysa yeniden tam yükleme yapın ve tam yüklemenin döndürmediği her nesneyi tablolarınızdan kaldırın.
• Haftalık tam yükleme. Besleme sorunsuz olsa bile haftada bir yapılan ve tablolarınızla karşılaştırılan bir tam yükleme, kopyanın eksiksiz olduğunu gösteren ucuz bir denetimdir.
• Anahtarın erişemediği veriler dışa aktarılmaz. İş, tam olarak anahtarının erişebildiği verilere ulaşır. Bir kart anahtarın erişiminden çıkarsa, örneğin erişimi kısıtlı bir koleksiyona taşındığı için, tam yükleme onu döndürmez; kartı kaldırılmış olarak ele alın.
• Sorular. Soru kaydının kendi dışa aktarması vardır: GET /v1/questions/export.csv bir dönemin sorularını puanlar ve kaynaklarla birlikte tek bir CSV dosyası olarak döndürür. Bunun için ask kapsamı gerekir.

Sınırlara uyma

Büyük bir kuruluşun tam yüklemesi çok sayıda istek yapar. Listeleri limit=200 ile birer birer okuyun ve RateLimit başlıklarını izleyin. 15 dakikada bir çalışan ve yalnızca değişiklikleri okuyan bir iş, her kotanın çok altında kalır.