Rates into your data warehouse

One Python script starts a run, waits for it to finish and saves every row as newline-delimited JSON, ready to load into your warehouse. Run it hourly, daily or weekly from your scheduler, or by hand whenever you need fresh rates.

In short

A Python script starts a run, waits for it to finish and saves every room and rate to files you can load into your warehouse. Your scheduler runs it as often as you like. It suits data teams, and each run costs one credit per property.

At a glance

Who it's for Data teams who keep rates in their own warehouse, such as BigQuery, Snowflake or PostgreSQL.
What you end up with Every room and rate for every property and check-in date, one table row each, plus one row per run, refreshed as often as you run the script.
What it costs One credit per property for every run, so how often you run it sets the cost. With 10 properties, a weekly run costs 10 credits a week, a daily run 70 and an hourly run 1,680. Every run checks all 365 check-in dates. Reading, retrying failed dates and deleting rows are free.
What you need A key with the Read and write scope, Python 3.8 or later with requests, a scheduler such as cron, and a warehouse to load into.

Steps

  1. Create a read and write key

    In Beacon Pro, open your initials in the top right, choose Company, then the API tab. Create a key with the Read and write scope, because starting a run spends credits. Its own runs is enough for its visibility, because the script only reads the runs it starts. If you ever replace the key, switch to the new one after the current period's run has finished. A key with Its own runs can't see runs another key started. See API keys.

    Store the key in your scheduler's secrets, and make it available to the script as the environment variable BEACON_KEY.

  2. Save the script

    Save this as scheduled_run.py. Change REFERENCE and REPORT to your own: a reference you'll recognise, a report name, a currency, a stay length in nights and property URLs (up to your company's limit, 50 by default). Set SCHEDULE to how often your scheduler will run it: "hourly", "daily" or "weekly". If you'll run it by hand, as and when, use "manual". 

    """Start a Rate Capture run, wait for it, and save it as ndjson files.

Run it from cron, Task Scheduler or your orchestrator as often as you want
fresh rates, or by hand whenever you need them. Set SCHEDULE to match.
Needs Python 3.8 or later and requests (pip install requests).
"""
import datetime
import json
import os
import sys
import time

import requests

API = "https://api.pro.getbeacon.cloud/v1"
REFERENCE = "harbour-competitors"
# How often your scheduler runs this script: "hourly", "daily" or "weekly".
# Use "manual" if you run it by hand, as and when: every run then starts a new capture.
SCHEDULE = "daily"
REPORT = {
    "name": "Harbour competitors",
    "currency": "GBP",
    "nights": 1,
    "properties": [
        {"url": "https://www.booking.com/hotel/gb/harbour-view.html", "name": "Harbour View Hotel"},
        {"url": "https://www.booking.com/hotel/gb/quayside-inn.html", "name": "Quayside Inn"},
    ],
}

session = requests.Session()
session.headers["Authorization"] = "Bearer " + os.environ["BEACON_KEY"]


class BeaconError(Exception):
    def __init__(self, message, code=None):
        super().__init__(message)
        self.code = code


def call(method, path, **kwargs):
    """Send one request, waiting and sending it again when the answer says to."""
    for attempt in range(12):
        try:
            r = session.request(method, API + path, timeout=300, **kwargs)
        except requests.RequestException as error:
            # Dropped connection or timeout: safe to send again, with the same Idempotency-Key on a write
            print(f"{method} {path}: {error.__class__.__name__}, trying again in 30 seconds")
            time.sleep(30)
            continue
        if r.ok:
            return r
        try:
            error = r.json()["error"]
        except (ValueError, KeyError):
            error = {"code": str(r.status_code), "message": r.text}
        code = error["code"]
        if code in ("rate_limited", "idempotency_in_progress"):
            time.sleep(int(r.headers.get("Retry-After", "60")))
        elif code == "too_many_active_runs":
            time.sleep(300)  # your company's limit of runs at once is reached: try again later
        elif code == "internal_error":
            time.sleep(30)  # safe to send again, with the same Idempotency-Key on a write
        else:
            raise BeaconError(f"{method} {path}: {r.status_code} {code}: {error['message']}", code)
    raise BeaconError(f"{method} {path}: gave up after {attempt + 1} tries")


def this_period(now):
    """When the current hour, day or week began (UTC), and a label for it."""
    if SCHEDULE == "hourly":
        start = now.replace(minute=0, second=0, microsecond=0)
        return start, start.strftime("%Y-%m-%dT%H")
    if SCHEDULE == "daily":
        start = now.replace(hour=0, minute=0, second=0, microsecond=0)
        return start, start.strftime("%Y-%m-%d")
    if SCHEDULE == "weekly":
        start = now.replace(hour=0, minute=0, second=0, microsecond=0) - datetime.timedelta(days=now.weekday())
        year, week, _ = start.isocalendar()
        return start, f"{year}-W{week:02d}"
    if SCHEDULE == "manual":
        return None, now.strftime("%Y-%m-%dT%H%M%S")
    raise BeaconError(f"SCHEDULE must be hourly, daily, weekly or manual, not {SCHEDULE!r}")


def find_run_since(start):
    """The newest run with this reference started since start, or None."""
    params = {"reference": REFERENCE, "from": start.date().isoformat(), "limit": 1}
    runs = call("GET", "/runs", params=params).json()["runs"]
    if not runs:
        return None
    run = runs[0]
    started = run["startedAt"]
    if started is None:  # not started yet, so it belongs to this period
        return run
    started = datetime.datetime.fromisoformat(started.replace("Z", "+00:00"))
    return run if started >= start else None


def start_run(idempotency_key):
    body = {"reference": REFERENCE, **REPORT}
    run = call("POST", "/runs", json=body, headers={"Idempotency-Key": idempotency_key}).json()
    print(f"Started run {run['id']} for {run['creditsCharged']} credits")
    return run


def wait_until_complete(run):
    while not run["complete"]:
        time.sleep(60)  # no more than once a minute
        run = call("GET", f"/runs/{run['id']}").json()
        print(f"Run {run['id']}: {run['status']}, {run['remaining']} checks left")
    return run


def retry_failed_once(run):
    """Run failed checks again, for free. Skipped on a 409: the run was retried in the last 10 minutes, or is running again."""
    if run["failed"] == 0:
        return run
    key = f"retry-{run['id']}-{run['finishedAt']}"  # one retry each time the run finishes
    try:
        answer = call("POST", f"/runs/{run['id']}/retry", headers={"Idempotency-Key": key}).json()
    except BeaconError as error:
        if error.code in ("retry_cooldown", "run_not_completed"):
            print(f"Run {run['id']}: not retried ({error.code}), keeping what it has")
            return run
        raise
    print(f"Run {run['id']}: retrying {answer['requeued']} failed checks")
    return wait_until_complete(answer["run"])


def download(run_id):
    """Write the rows to one file and the run to another, named by run id.

    Each file is written under a temporary name and renamed when complete, and the
    run file goes last, so a run file means its rows file is whole.
    """
    rows_file = f"beacon-rows-{run_id}.ndjson"
    run_file = f"beacon-run-{run_id}.ndjson"
    r = call("GET", f"/runs/{run_id}/results", params={"format": "ndjson"}, stream=True)
    lines = (line for line in r.iter_lines() if line)
    run_line = json.loads(next(lines))  # the first line is the run, with "kind": "run"
    run_line.pop("kind", None)
    count = 0
    with open(rows_file + ".part", "w", encoding="utf-8", newline="\n") as out:
        for line in lines:
            row = json.loads(line)
            row["runId"] = run_id  # rows don't carry the run id, so add it
            out.write(json.dumps(row, ensure_ascii=False) + "\n")
            count += 1
    os.replace(rows_file + ".part", rows_file)
    with open(run_file + ".part", "w", encoding="utf-8", newline="\n") as out:
        out.write(json.dumps(run_line, ensure_ascii=False) + "\n")
    os.replace(run_file + ".part", run_file)
    print(f"Run {run_id}: {count} rows in {rows_file}, run in {run_file}")


def main():
    start, label = this_period(datetime.datetime.now(datetime.timezone.utc))

    run = find_run_since(start) if start else None
    if run is None:
        run = start_run(f"{REFERENCE}-{label}")
    elif run["resultsPurgedAt"] is not None:
        print(f"Run {run['id']} is already stored and its rows deleted")
        return

    run = wait_until_complete(run)
    run = retry_failed_once(run)
    download(run["id"])


if __name__ == "__main__":
    try:
        main()
    except BeaconError as error:
        sys.exit(str(error))

    What it does, in order:

    • Looks for a run from this hour, day or week first. If a run with your reference has already started in the current period (in UTC), the script picks it up instead of starting another. So a second run of the script in the same period costs nothing extra. With "manual", every run of the script starts a new capture.
    • Starts the run with POST /v1/runs and an Idempotency-Key made of the reference and the period, such as harbour-competitors-2026-10-08 for a daily run. If the connection drops, the script sends the request again with the same key. Beacon then answers with the first run instead of starting and charging a second one. Beacon forgets idempotency keys after 24 hours, which is why the script also does the check above.
    • Waits for it to finish, asking for the run once a minute until complete is true. A run usually takes minutes rather than seconds.
    • Retries failed checks once, which is free. If the run was retried in the last 10 minutes, for example from the app, the script keeps what it has.
    • Downloads the whole run in one request with format=ndjson. The first line is the run, marked "kind": "run", and every line after it is one row.
    • Writes two files named by run id: beacon-rows-1402.ndjson with one row per line, and beacon-run-1402.ndjson with the run. Rows don't carry the run's id, so the script adds it to each row as runId. Each file is written under a temporary name and renamed when it's complete, and the run file goes last, so if the run file is there, its rows file is whole.
    • Waits when Beacon asks it to. On 429 rate_limited and 409 idempotency_in_progress it waits for Retry-After. On 429 too_many_active_runs it tries again every 5 minutes. On 500 internal_error or a dropped connection, it waits 30 seconds and sends the request again. It gives up after 12 tries. Any other error stops the script with the error's code and message. The exit code is not zero, so your scheduler can alert on it.

    To see what the rows look like before you run anything, download the sample results file. Every column is described in field dictionary.

  3. Schedule it, or run it by hand

    Runs don't repeat by themselves, so your scheduler starts each one. Pick the line that matches your SCHEDULE. These are for cron; Task Scheduler and orchestrators such as Airflow take the same timings. 

    # Hourly, at 5 minutes past each hour
5 * * * *  cd /opt/beacon && /usr/bin/python3 scheduled_run.py >> scheduled_run.log 2>&1

# Daily, at 06:00
0 6 * * *  cd /opt/beacon && /usr/bin/python3 scheduled_run.py >> scheduled_run.log 2>&1

# Weekly, on Mondays at 06:00
0 6 * * 1  cd /opt/beacon && /usr/bin/python3 scheduled_run.py >> scheduled_run.log 2>&1

    To run it by hand instead, set SCHEDULE = "manual" and run python3 scheduled_run.py whenever you want fresh rates.

    The script can run for a while. A run usually takes minutes, but the more properties it has, the longer it takes, and it waits longer still if your company already has as many runs going as it may. Give the job at least an hour before your scheduler times it out.

  4. Load the files into your warehouse

    In BigQuery, the first load can detect the columns for you. Load the rows and the run into two tables:

    bq mk --dataset beacon
bq load --source_format=NEWLINE_DELIMITED_JSON --autodetect beacon.rate_rows beacon-rows-1402.ndjson
bq load --source_format=NEWLINE_DELIMITED_JSON --autodetect beacon.runs beacon-run-1402.ndjson

    Detection only looks at the first rows of a file, and many fields, such as roomsLeft and undercutBy, are null on most rows. So save the detected schema and check it once against the field dictionary. BigQuery can't change a column's type in place, so if one is wrong, correct it in the saved file, delete the table and load the first file again with that schema instead of --autodetect: 

    bq show --schema --format=prettyjson beacon.rate_rows > rate_rows_schema.json
# correct any column types in rate_rows_schema.json, then:
bq rm -f -t beacon.rate_rows
bq load --source_format=NEWLINE_DELIMITED_JSON beacon.rate_rows beacon-rows-1402.ndjson rate_rows_schema.json

    Then load every later run with the same schema. Deleting the run's rows first means loading the same file twice never doubles it:

    bq query --use_legacy_sql=false 'DELETE FROM beacon.rate_rows WHERE runId = 1403'
bq load --source_format=NEWLINE_DELIMITED_JSON beacon.rate_rows beacon-rows-1403.ndjson rate_rows_schema.json

    Do the same for the runs table, keyed on id. In Snowflake, COPY INTO a table with one VARIANT column, using FILE_FORMAT = (TYPE = JSON), loads one row per line. In PostgreSQL, insert each line into a jsonb column, for example from the same script.

  5. Optional: delete Beacon's copy

    Once the rows are safely in your warehouse, you can tell Beacon to delete its copy. The run, its counters and its cost stay. Only do this after the load has succeeded: afterwards the run's results and rates answer 410 results_purged, and its failed dates can't be retried. 

    curl -X DELETE https://api.pro.getbeacon.cloud/v1/runs/1402/results \
  -H "Authorization: Bearer $BEACON_KEY" \
  -H "Idempotency-Key: delete-1402"
    This also changes what the Lighthouse-compatible API reads

    A Lighthouse subscription reads the newest completed run with that reference whose rows still exist. Once you delete a run's rows, any spreadsheet, dashboard or tool reading that reference falls back to an older run, or gets no rates at all. If anyone reads your reference that way, keep the rows. See Switching from Lighthouse.

What to watch for

  • Runs at once. A company can have 5 runs queued or running at the same time, unless Beacon has raised its limit. Another start answers 429 too_many_active_runs, with no Retry-After, and charges nothing. If you start runs for many references, start them one after another, or space them out.
  • Not enough credits. If a start is refused with 402 insufficient_credits, the same Idempotency-Key gets the same answer for 24 hours. After buying credits, run the script once with SCHEDULE = "manual", which uses a new key, or wait for the next period.
  • Failed checks. A completed run with failed above 0 is missing those checks, and their rows say not fetched. Retrying is free. You can retry a run once every 10 minutes, counting retries from the API and from the app. If someone retried the run in the app just before, the script's own retry answers 409 retry_cooldown. The script expects that and loads what it has.
  • Key scope. Starting runs, retrying and deleting need a read_write key. A read key answers 403 scope_read_only.
  • Placeholder rows. A date with no rooms still has a row, with availability set to sold out, not fetched, past, not fetched yet or min stay N nights and every room field null. Filter on availability = 'available' for prices, and count the others as days without a rate, not as a price of 0.
  • Duplicates. Each run is a separate snapshot, and the same check-in date appears in many runs. Keep runId on every row, delete a run's rows before loading it again, and choose a run in your queries rather than mixing them.
  • Time zones. Every time, such as startedAt, is UTC with a Z, and the from filter on GET /v1/runs is a UTC date. Check-in dates are plain dates, and a run's dates start on today's date in its timeZone. So dateFrom can be a day before or after the date in startedAt.
  • Retention. Your company may have a retention period after which rows are deleted. Download each run soon after it completes.
Field dictionaryEvery column in the rows and the run. How a run progressesStatuses, counters and failed checks. Quick startThe same calls, one at a time. ErrorsEvery error code and what to do about it.