증분 동기화

import os

import requests

BASE = "https://tahoe.workonward.com/api/partner/v1"
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['TAHOE_API_KEY']}"


def current_sequence():
    """The sequence of the newest event this key can see, or None if none."""
    after, last = None, None
    while True:
        params = {"limit": 100}
        if after is not None:
            params["after"] = after
        response = session.get(f"{BASE}/events", params=params, timeout=30)
        response.raise_for_status()
        body = response.json()
        if body["data"]:
            last = body["data"][-1]["sequence"]
        if not body["has_more"]:
            return last
        after = body["next_after"]


def cold_start(store):
    # 1. The watermark FIRST, held in memory for now.
    watermark = current_sequence()

    # 2. The backfill. Slow, and that is fine: it happens once.
    backfill_jobs(store)
    backfill_applicants(store)
    backfill_applications(store)

    # 3. Save the watermark only now. A crash during the backfill then
    #    restarts the whole cold start, instead of resuming from a point
    #    the copy never reached.
    store.save_sequence(watermark)

def drain(store):
    after = store.load_sequence()

    while True:
        params = {"limit": 100}
        if after is not None:
            params["after"] = after
        response = session.get(f"{BASE}/events", params=params, timeout=30)
        response.raise_for_status()
        body = response.json()

        for event in body["data"]:
            # At-least-once delivery: the same event can arrive more than
            # once, so this check is not an optimization.
            if not store.already_processed(event["id"]):
                dispatch(store, event)
                store.mark_processed(event["id"])
            # AFTER the event is handled. Saving first turns a crash halfway
            # through a page into events that are skipped for good. Use the
            # event's own sequence: next_after is null on the last page.
            after = event["sequence"]
            store.save_sequence(after)

        if not body["has_more"]:
            return

# This system records outcomes only, not pipeline stages.
MIRRORED_APPLICATION_FIELDS = {"status"}


def on_application_event(store, event):
    changed = event["data"]["object"].get("changed")
    # When the event lists what changed and you copy none of it, skip the
    # read. When there is no list, read again to be safe.
    if changed and not set(changed) & MIRRORED_APPLICATION_FIELDS:
        return

    handle = event["data"]["object"]["id"]
    response = session.get(f"{BASE}/applications/{handle}", timeout=30)
    if response.status_code == 404:
        store.delete_application(handle)  # gone since the event was written
        return
    response.raise_for_status()
    store.upsert_application(response.json())