Ekspor ke gudang data

Analis sering menginginkan data Sursa berada di samping data perusahaan lainnya: kartu mana yang terverifikasi dan oleh siapa, bagaimana pengetahuan bertambah per tim, pertanyaan mana yang masih belum terjawab. Panduan ini membangun job terjadwal yang menyalin Sursa ke gudang data Anda dan menjaga salinan itu tetap terbaru.

Job ini menggabungkan tiga bagian API:

Anda membutuhkan kunci dengan cakupan cards:read dan users:read, ditambah webhooks:manage untuk umpan peristiwa. Berikan job ini kuncinya sendiri dengan peran admin: daftar lengkap pengguna dan umpan peristiwa hanya terbuka untuk peran itu. Kunci dengan peran dan profil data yang lebih sempit hanya mengekspor apa yang boleh dilihat kelompok tertentu; kunci itu membaca daftar ringkas pengguna (brief=true) dan tidak membaca umpan peristiwa, jadi penghapusan diambilnya dari muatan penuh mingguan di bawah.

Tabel

Tata letak sederhana dengan satu tabel per objek dan satu tabel untuk peristiwa:

Simpan setiap objek apa adanya, sebagai kolom JSON, ditambah kolom yang sering Anda kueri (updated_at, collection_id, status, owner_id). Field baru muncul di versi API berikutnya; dengan kolom JSON, Anda mendapatkannya tanpa mengubah job.

Muatan pertama

Baca setiap daftar sepenuhnya, tanpa updated_since, dan tulis setiap objek. Ingat saat muatan itu dimulai: eksekusi berikutnya berlanjut dari sana.

Muatan inkremental

Pada setiap eksekusi, misalnya setiap 15 menit:

1. Ingat waktu mulai eksekusi ini.
2. Baca kartu dengan updated_since diisi waktu mulai eksekusi sebelumnya, dikurangi cadangan waktu satu menit, dan minta setiap halaman berikutnya dengan kursor dan updated_since yang sama (kursor tanpa penyaring halaman pertama mengembalikan 400 api.cursor_invalid). Baca koleksi, pengguna, dan grup secara utuh: koleksi dan pengguna tidak memiliki updated_at, dan updated_at grup tidak berubah saat anggotanya berubah. Buat atau ganti setiap objek menurut id.
3. Baca umpan peristiwa dari kursor yang Anda simpan terakhir kali. Tambahkan setiap peristiwa ke sursa_events dan terapkan penghapusan: peristiwa card.deleted menghapus kartu dari sursa_cards, atau menandainya sebagai terhapus jika Anda menyimpan riwayat.
4. Simpan waktu mulai dan kursor umpan yang baru, dalam transaksi yang sama dengan data jika gudang data Anda mendukungnya.

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

Kursor umpan dikembalikan di setiap halaman, bahkan di halaman terakhir. Dengan menyimpannya, eksekusi berikutnya dimulai tepat setelah peristiwa terakhir yang Anda simpan, jadi tidak ada peristiwa yang dibaca dua kali atau terlewat.

Menjaga salinan tetap akurat

• Lebih dari 30 hari tanpa eksekusi. Umpan menyimpan peristiwa selama 30 hari. Jika job tidak berjalan lebih lama dari itu, jalankan lagi muatan penuh, dan hapus dari tabel Anda setiap objek yang tidak dikembalikan oleh muatan penuh itu.
• Muatan penuh mingguan. Bahkan dengan umpan yang sehat, muatan penuh seminggu sekali yang dibandingkan dengan tabel Anda adalah pemeriksaan murah bahwa salinannya lengkap.
• Apa yang tidak dapat dilihat kunci tidak diekspor. Job melihat persis apa yang dilihat kuncinya. Jika sebuah kartu keluar dari akses kunci, misalnya karena dipindahkan ke koleksi terbatas, muatan penuh tidak akan mengembalikannya; tangani kartu itu sebagai terhapus.
• Pertanyaan. Log pertanyaan memiliki ekspornya sendiri: GET /v1/questions/export.csv mengembalikan pertanyaan dari suatu periode sebagai satu file CSV, beserta penilaian dan sumbernya. Ekspor ini memerlukan cakupan ask.

Menghormati batas

Muatan penuh untuk organisasi besar mengirim banyak permintaan. Baca satu daftar dalam satu waktu dengan limit=200, dan perhatikan header RateLimit. Job setiap 15 menit yang hanya membaca perubahan tetap jauh di bawah kuota apa pun.