Eksports uz datu noliktavu

Analītiķi bieži vēlas Sursas datus blakus pārējiem uzņēmuma datiem: kuras kartītes ir pārbaudītas un kas tās pārbaudījis, kā zināšanas pieaug katrā komandā, kuri jautājumi paliek neatbildēti. Šajā ceļvedī tiek izveidots ieplānots darbs, kas kopē Sursas datus jūsu datu noliktavā un uztur kopiju aktuālu.

Darbs apvieno trīs API daļas:

Jums vajadzīga atslēga ar tvērumiem cards:read un users:read, kā arī webhooks:manage notikumu plūsmai. Piešķiriet darbam savu atslēgu ar lomu admin: pilns lietotāju saraksts un notikumu plūsma ir pieejami tikai tai. Atslēga ar šaurāku lomu un datu profilu eksportē tikai to, ko drīkst redzēt konkrēta grupa; tā lasa īso lietotāju sarakstu (brief=true) un nelasa notikumu plūsmu, tāpēc dzēstos objektus tā uzzina no tālāk aprakstītās iknedēļas pilnās ielādes.

Tabulas

Vienkāršs izkārtojums – viena tabula katram objektam un viena tabula notikumiem:

Glabājiet katru objektu tādu, kāds tas pienāk, JSON kolonnā, kā arī kolonnas, kuras bieži izmantojat vaicājumos (updated_at, collection_id, status, owner_id). Nākamajās API versijās parādās jauni lauki; ar JSON kolonnu jūs tos saņemat, nemainot darbu.

Pirmā ielāde

Izlasiet katru sarakstu pilnībā, bez updated_since, un ierakstiet katru objektu. Atcerieties brīdi, kad ielāde sākās: nākamā palaišana turpinās no tā.

Inkrementālas ielādes

Katrā palaišanā, piemēram, ik pēc 15 minūtēm:

1. Atcerieties šīs palaišanas sākuma laiku.
2. Kartītes lasiet ar updated_since, kas iestatīts uz iepriekšējās palaišanas sākuma laiku, atskaitot minūti rezervei, un katru nākamo lapu pieprasiet ar kursoru un to pašu updated_since (kursors bez pirmās lapas filtriem atgriež 400 api.cursor_invalid). Kolekcijas, lietotājus un grupas lasiet pilnībā: kolekcijām un lietotājiem nav updated_at, bet grupas updated_at nemainās, kad mainās tās dalībnieki. Katru objektu izveidojiet vai aizstājiet pēc id.
3. Lasiet notikumu plūsmu no kursora, ko saglabājāt iepriekšējo reizi. Pievienojiet katru notikumu tabulai sursa_events un piemērojiet dzēšanu: notikums card.deleted noņem kartīti no sursa_cards vai atzīmē to kā dzēstu, ja glabājat vēsturi.
4. Saglabājiet sākuma laiku un jauno plūsmas kursoru tajā pašā transakcijā, kurā datus, ja jūsu noliktava to atbalsta.

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

Plūsmas kursors tiek atgriezts katrā lapā, arī pēdējā. Ja to saglabājat, nākamā palaišana sākas tieši aiz pēdējā saglabātā notikuma, tāpēc neviens notikums netiek nolasīts divreiz vai izlaists.

Kā uzturēt kopiju precīzu

• Vairāk nekā 30 dienas bez palaišanas. Plūsma glabā notikumus 30 dienas. Ja darbs nav palaists ilgāk, vēlreiz veiciet pilnu ielādi un izdzēsiet no tabulām katru objektu, ko pilnā ielāde neatgrieza.
• Iknedēļas pilna ielāde. Pat ja plūsma darbojas bez traucējumiem, pilna ielāde reizi nedēļā, salīdzināta ar jūsu tabulām, ir lēta pārbaude, ka kopija ir pilnīga.
• Ko atslēga neredz, tas netiek eksportēts. Darbs redz tieši to, ko redz tā atslēga. Ja kartīte iziet ārpus atslēgas piekļuves, piemēram, tāpēc, ka tā pārvietota uz ierobežotu kolekciju, pilnā ielāde to neatgriezīs; apstrādājiet to kā noņemtu.
• Jautājumi. Jautājumu žurnālam ir savs eksports: GET /v1/questions/export.csv atgriež perioda jautājumus vienā CSV failā ar vērtējumiem un avotiem. Tam vajadzīgs tvērums ask.

Ievērojiet limitus

Lielas organizācijas pilnā ielāde veic daudz pieprasījumu. Lasiet vienu sarakstu vienlaikus ar limit=200 un sekojiet galvenēm RateLimit. Darbs ik pēc 15 minūtēm, kas lasa tikai izmaiņas, paliek tālu zem jebkuras kvotas.