Sync rows on a schedule

A nightly job that copies a day's rows out of Halo with keyset paging, stays inside the row budget, resumes where it stopped, and keeps working as its key is rolled.

This job copies yesterday's orders by region and channel into a local table each night. It pages by keyset, saves its place after every window, catches up if a night stopped early, and can be run again safely. It needs Python 3.11 or newer.

Plan it

Choose what a row is. Results are grouped, so your dimensions decide the grain of the table you're filling. Use dimensions that are unique together, and make them your table's key.

Size it against the budget. A person's keys may read a set number of rows in a rolling 24 hours: 100,000 on Standard, 500,000 on Professional and 2,000,000 on Enterprise. Tonight's rows are still counted tomorrow night if the job runs a little earlier, so aim well under it. A job that needs more should aggregate more or run less of the history at a time. See The daily row budget.

Use absolute dates. A resumed run must ask the same question, so save the day you're on. yesterday would mean a different day tomorrow.

Name the time zone. Set timeZone, so the job doesn't move if someone changes the organization's.

Use a narrow key. data.query is all this needs: whoami takes no scope. The person's role needs data.export, and Marketplace data can't be copied: 403 marketplace_data_not_available.

The job

import datetime
import json
import os
import sqlite3
import sys
import time
import zoneinfo

import requests

API = "https://api.usehalo.net/v1/developer"
DATASET_ID = os.environ["HALO_DATASET_ID"]
TIME_ZONE = "America/Denver"

session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['HALO_API_KEY']}"

db = sqlite3.connect("orders.db")
db.executescript("""
    create table if not exists orders_daily (
        day text, region text, channel text, revenue real, orders integer,
        primary key (day, region, channel)
    );
    create table if not exists sync_state (
        id integer primary key check (id = 1), day text, next_offset integer, after text
    );
""")


class BudgetSpent(Exception):
    pass


def call(method, path, **kwargs):
    for attempt in range(1, 6):
        response = session.request(method, f"{API}{path}", timeout=330, **kwargs)
        if response.ok:
            return response.json()
        problem = response.json()
        code = problem.get("code")
        if code == "row_budget_exceeded":
            raise BudgetSpent(problem.get("detail"))
        if code not in {"rate_limited", "query_busy", "source_error", "request_timeout", "internal"}:
            sys.exit(f"{code}: {problem.get('detail')} (request {problem.get('requestId')})")
        time.sleep(float(response.headers.get("Retry-After", 2**attempt)))
    sys.exit(f"{code}: gave up after 5 tries (request {problem.get('requestId')})")


def yesterday():
    today = datetime.datetime.now(zoneinfo.ZoneInfo(TIME_ZONE)).date()
    return today - datetime.timedelta(days=1)


def save_place(day, offset, after):
    db.execute("insert or replace into sync_state values (1, ?, ?, ?)", (day, offset, after))
    db.commit()


def copy_day(day, offset, after):
    query = {
        "dimensions": [{"field": "order_date"}, {"field": "region"}, {"field": "channel"}],
        "measures": [{"field": "revenue"}, {"field": "orders"}],
        "filters": [
            {
                "field": "order_date",
                "op": "date_range",
                "range": {"kind": "absolute", "start": day, "end": day},
            }
        ],
        "limit": 1000,
        "keyset": True,
        "timeZone": TIME_ZONE,
    }
    while True:
        window = {**query, "offset": offset}
        if after:
            window["cursor"] = after
        page = call("POST", f"/datasets/{DATASET_ID}/query", json=window)
        stats = page["stats"]

        # Saving the rows and the place together means a crash can't skip any.
        db.executemany("insert or replace into orders_daily values (?, ?, ?, ?, ?)", page["rows"])
        offset += stats["rowCount"]
        after = stats.get("nextCursor")
        if not stats["hasMore"]:
            next_day = datetime.date.fromisoformat(day) + datetime.timedelta(days=1)
            save_place(next_day.isoformat(), 0, None)
            print(f"{day}: copied {offset} rows")
            return
        save_place(day, offset, after)


def main():
    who = call("GET", "/whoami")
    ends = datetime.datetime.fromisoformat(who["key"]["expiresAt"])
    days_left = (ends - datetime.datetime.now(datetime.UTC)).days
    if days_left < 14:
        print(f"Warning: this key ends in {days_left} days. Roll it.", file=sys.stderr)
    print(f"{who['limits']['rowBudget']['left']} rows left in the budget")

    # The place is a day and a window, so a night that stopped early is caught up.
    saved = db.execute("select day, next_offset, after from sync_state").fetchone()
    day, offset, after = saved or (yesterday().isoformat(), 0, None)
    try:
        while datetime.date.fromisoformat(day) <= yesterday():
            copy_day(day, offset, after)
            day, offset, after = db.execute("select day, next_offset, after from sync_state").fetchone()
    except BudgetSpent as spent:
        print(f"Stopping, and resuming on the next run: {spent}", file=sys.stderr)
        return 75
    return 0


sys.exit(main())

The columns come back in the order the query names them, so page["rows"] fits the table as it is. A real job should check page["columns"] against what it expects, so a renamed field stops the job rather than filling the wrong column.

15 2 * * * cd /opt/halo-sync && python sync_orders.py >> sync.log 2>&1

Give cron the key from a secret store or a root-only environment file, rather than writing it in the crontab.

What happens when

A resumed run uses its saved cursor if nothing has changed. If the dataset's definition has, Halo ignores the cursor and uses offset, so the run still finishes, and writing by primary key means any row that moved is only overwritten. Rows that changed at the source during a run can be missed or repeat, so compare against a later run if you need certainty.

Check on success as well as failure. Alert if a night passes with no copied line, since a job that never starts writes nothing to its log. To see what a day will cost before you read it, ask for countRows: true on its first window. It runs a second query, and the count lands in stats.total.

Roll the key

Keys end: after 90 days by default, and at most 365. The job warns from 14 days out, and Halo tells the person seven days before. A key can't roll itself, so a person does this at the keys page:

1. Roll the key. The old one keeps working for 24 hours.

2. Update your secret store with the new secret.

3. Run the job once by hand, or wait for tonight's run. key.id in whoami is now the new key's.

Both keys count against the same person's budget, and the budget doesn't reset when you roll.

When its person leaves

A key stops when the person who made it changes their password or two-factor, or leaves the organization. For a job, that can happen with no warning to you, so treat 401 invalid_api_key as an alert, and know who to ask for a new key. Keys owned by the organization are planned.