Εξαγωγή σε αποθήκη δεδομένων

Οι αναλυτές συχνά θέλουν τα δεδομένα της 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 λεπτά που διαβάζει μόνο τις αλλαγές μένει πολύ κάτω από οποιαδήποτε ποσόστωση.