Xuất sang kho dữ liệu

Các nhà phân tích thường muốn dữ liệu Sursa nằm cạnh phần còn lại của dữ liệu công ty: những thẻ nào đã được kiểm chứng và bởi ai, tri thức tăng lên như thế nào theo từng đội, những câu hỏi nào vẫn chưa có câu trả lời. Hướng dẫn này xây dựng một tác vụ chạy theo lịch, sao chép Sursa vào kho dữ liệu của bạn và giữ bản sao luôn cập nhật.

Tác vụ kết hợp ba phần của API:

Bạn cần một khóa có phạm vi quyền cards:read và users:read, cùng webhooks:manage cho nguồn cấp sự kiện. Hãy cấp cho tác vụ một khóa riêng có vai trò admin: danh sách đầy đủ người dùng và nguồn cấp sự kiện chỉ mở cho vai trò này. Khóa có vai trò và hồ sơ dữ liệu hẹp hơn chỉ xuất những gì một nhóm cụ thể được xem; nó đọc danh sách rút gọn người dùng (brief=true) và không đọc nguồn cấp sự kiện, nên nó lấy các thao tác xóa từ lần nạp toàn bộ hằng tuần ở bên dưới.

Bảng

Một cách bố trí đơn giản với một bảng cho mỗi đối tượng và một bảng cho sự kiện:

Hãy lưu mỗi đối tượng đúng như khi nhận được, trong một cột JSON, cộng thêm các cột bạn thường truy vấn (updated_at, collection_id, status, owner_id). Trường mới xuất hiện trong các phiên bản API sau; nhờ cột JSON, bạn nhận được chúng mà không cần thay đổi tác vụ.

Lần nạp đầu

Đọc toàn bộ từng danh sách, không có updated_since, và ghi mọi đối tượng. Ghi nhớ thời điểm lần nạp bắt đầu: lần chạy tiếp theo tiếp tục từ đó.

Các lần nạp gia tăng

Ở mỗi lần chạy, ví dụ 15 phút một lần:

1. Ghi nhớ thời điểm bắt đầu của lần chạy này.
2. Đọc thẻ với updated_since đặt bằng thời điểm bắt đầu của lần chạy trước, trừ đi một phút dự phòng, và yêu cầu mỗi trang tiếp theo bằng con trỏ cùng giá trị updated_since đó (con trỏ không kèm các bộ lọc của trang đầu tiên trả về 400 api.cursor_invalid). Đọc toàn bộ bộ sưu tập, người dùng và nhóm: bộ sưu tập và người dùng không có updated_at, còn updated_at của nhóm không thay đổi khi thành viên của nhóm thay đổi. Tạo hoặc thay thế từng đối tượng theo id.
3. Đọc nguồn cấp sự kiện từ con trỏ bạn đã lưu lần trước. Thêm từng sự kiện vào sursa_events và áp dụng các thao tác xóa: sự kiện card.deleted xóa thẻ khỏi sursa_cards, hoặc đánh dấu thẻ đã bị xóa nếu bạn giữ lịch sử.
4. Lưu thời điểm bắt đầu và con trỏ mới của nguồn cấp, trong cùng giao dịch với dữ liệu nếu kho dữ liệu của bạn hỗ trợ.

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

Con trỏ của nguồn cấp được trả về trên mọi trang, kể cả trang cuối cùng. Lưu nó nghĩa là lần chạy tiếp theo bắt đầu ngay sau sự kiện cuối cùng bạn đã lưu, nên không sự kiện nào bị đọc hai lần hay bị bỏ sót.

Giữ bản sao chính xác

• Hơn 30 ngày không chạy. Nguồn cấp lưu giữ sự kiện trong 30 ngày. Nếu tác vụ không chạy lâu hơn khoảng đó, hãy chạy lại lần nạp toàn bộ, và xóa khỏi các bảng của bạn mọi đối tượng mà lần nạp toàn bộ không trả về.
• Nạp toàn bộ hằng tuần. Ngay cả khi nguồn cấp hoạt động tốt, một lần nạp toàn bộ mỗi tuần, đối chiếu với các bảng của bạn, là cách kiểm tra ít tốn kém rằng bản sao đầy đủ.
• Những gì khóa không xem được sẽ không được xuất. Tác vụ nhìn thấy đúng những gì khóa của nó nhìn thấy. Nếu một thẻ ra khỏi phạm vi truy cập của khóa, ví dụ vì nó được chuyển vào một bộ sưu tập bị hạn chế, lần nạp toàn bộ sẽ không trả về nó; hãy xử lý nó như đã bị xóa.
• Câu hỏi. Nhật ký câu hỏi có cách xuất riêng: GET /v1/questions/export.csv trả về các câu hỏi của một khoảng thời gian trong một tệp CSV, kèm đánh giá và nguồn. Endpoint này cần phạm vi ask.

Tôn trọng các giới hạn

Lần nạp toàn bộ của một tổ chức lớn gửi nhiều yêu cầu. Hãy đọc từng danh sách một với limit=200, và theo dõi các header RateLimit. Một tác vụ chạy 15 phút một lần và chỉ đọc các thay đổi sẽ luôn nằm xa dưới mọi hạn mức.