Sync car auction data: an hourly import that doesn't break

How to sync car auction data: the shape of the job

Three requests make up the whole loop. The first runs once (and again whenever you reconcile); the other two run every hour, in the same job, one after the other.

Every request is a GET with your key in the x-api-key header and no body. Every authenticated request counts toward the plan allowance, including requests that end in an error, so validate parameters before you send them and leave unused ones out entirely: an empty domain_id= can be rejected.

A small client with retries and pagination

Start with the part everything else uses: one HTTP session, a request function that knows which failures are worth retrying, and a generator that walks pages. The collections from /cars and /archived-lots come in a data array with links and meta; you follow until links.next is null.

import jsonimport loggingimport mathimport osimport randomimport sqlite3import sysimport timefrom datetime import datetime, timezone import requests log = logging.getLogger("auction-sync") BASE = "https://auctionsapi.com/api/"API_KEY = os.environ.get("AUCTIONS_API_KEY", "YOUR_API_KEY")RETRYABLE = {429, 500, 502, 503, 504} session = requests.Session()session.headers.update({    "x-api-key": API_KEY,    "accept": "application/json",})  class SyncError(RuntimeError):    pass  def backoff(response, attempt):    """Honour Retry-After, otherwise exponential backoff with jitter."""    value = None    if response is not None:        value = response.headers.get("Retry-After")    if value and value.isdigit():        return min(int(value), 300)    return min(2 ** attempt + random.random(), 60)  def get_json(endpoint, params, attempts=5):    for attempt in range(attempts):        try:            r = session.get(BASE + endpoint, params=params,                            timeout=(10, 60), allow_redirects=False)        except (requests.Timeout, requests.ConnectionError):            r = None        if r is not None and r.status_code == 200:            try:                return r.json()            except ValueError:                r = None  # truncated body: treat like a network error        elif r is not None and r.status_code not in RETRYABLE:            # 400 bad parameter, 401 source not in the key, 403 key            # problem: sending the same request again will not help.            raise SyncError(f"{endpoint}: HTTP {r.status_code}")        time.sleep(backoff(r, attempt))    raise SyncError(f"{endpoint}: gave up after {attempts} attempts")  def pages(endpoint, params):    """Yield data[] page by page until links.next is null."""    page = 1    while True:        body = get_json(endpoint, {**params, "page": page})        if not isinstance(body, dict) or not isinstance(                body.get("data"), list):            raise SyncError(f"{endpoint}: unexpected envelope")        yield body["data"]        links = body.get("links")        if not isinstance(links, dict) or not links.get("next"):            return        page += 1  # same filters and per_page, only page changes

Two choices in this code are deliberate. The generator does not follow the next URL; it rebuilds the request with the same filters and only increments page. That keeps your filters identical on every page and means your key is never sent anywhere except the API host. And simple_paginate=1 skips the total count, which makes deep pages faster; do not build logic on meta.total anyway.

Page sizes per plan

/cars returns 50 vehicles per request by default on every plan. Demo and Small keys are capped at 50; Unlimited accepts per_page=1000. /archived-lots defaults to 100, with the same 1000 ceiling on Unlimited. A per_page above your plan's limit is not quietly reduced: the request fails with 400. Keep both sizes in configuration, which is what CARS_PER_PAGE and ARCHIVE_PER_PAGE are for in the last part of the script. The pagination page lists the sizes of every endpoint.

Upsert by lot, not by VIN

A vehicle is the physical car; a lot is one appearance of that car at an auction. A 2018 Honda CR-V can fail to sell at Copart, come back at IAAI a fortnight later and sell there, which gives one vehicle and two lots. The same lot number can also exist on both auctions. So:

• Key vehicles on vehicle.id.
• Key lots on lot.id. It is also the lot_id that the archive feed refers to, so it is your join key.
• Add a unique index on (domain, lot): the auction plus its own lot number is what people search for.
• Never upsert on the VIN alone. You would merge separate sales into one row and lose history.

SCHEMA = """CREATE TABLE IF NOT EXISTS vehicles (  id        INTEGER PRIMARY KEY,   -- vehicle.id  vin       TEXT,  payload   TEXT NOT NULL,         -- vehicle JSON without lots  synced_at TEXT NOT NULL);CREATE TABLE IF NOT EXISTS lots (  id              INTEGER PRIMARY KEY,  -- lot.id = lot_id in the archive  vehicle_id      INTEGER NOT NULL,  domain          TEXT,                 -- copart_com or iaai_com  lot             TEXT,                 -- the auction's lot number  archived        INTEGER NOT NULL DEFAULT 0,  archived_at     TEXT,  final_bid       REAL,  payload         TEXT,  archive_payload TEXT,  synced_at       TEXT NOT NULL);CREATE UNIQUE INDEX IF NOT EXISTS lots_source_lot ON lots(domain, lot);CREATE INDEX IF NOT EXISTS lots_vehicle ON lots(vehicle_id);CREATE TABLE IF NOT EXISTS sync_state (  name        TEXT PRIMARY KEY,  finished_at REAL NOT NULL);"""

The vehicle upsert stores the vehicle without its lots, then writes each lot. Note the WHERE lots.archived = 0 on the lot update: inventory changes while you page, so a page fetched a minute before a lot was archived must not flip it back to active.

def is_id(value):    return isinstance(value, int) and not isinstance(value, bool) \        and value > 0  def source_of(record):    domain = record.get("domain")    return domain.get("name") if isinstance(domain, dict) else None  def save_vehicle(db, car, now):    if not isinstance(car, dict) or not is_id(car.get("id")):        log.warning("skipped a malformed vehicle")        return    meta = {k: v for k, v in car.items() if k != "lots"}    db.execute(        """INSERT INTO vehicles (id, vin, payload, synced_at)           VALUES (?, ?, ?, ?)           ON CONFLICT(id) DO UPDATE SET vin = excluded.vin,             payload = excluded.payload,             synced_at = excluded.synced_at""",        (car["id"], car.get("vin"), json.dumps(meta), now))    lots = car.get("lots") if isinstance(car.get("lots"), list) else []    for lot in lots:        if not isinstance(lot, dict) or not is_id(lot.get("id")):            continue        db.execute(            """INSERT INTO lots (id, vehicle_id, domain, lot,                                 payload, synced_at)               VALUES (?, ?, ?, ?, ?, ?)               ON CONFLICT(id) DO UPDATE SET payload = excluded.payload,                 domain = excluded.domain, lot = excluded.lot,                 synced_at = excluded.synced_at               WHERE lots.archived = 0""",            (lot["id"], car["id"], source_of(lot), lot.get("lot"),             json.dumps(lot), now))

One more detail from the API contract: a domain_id filter selects vehicles that have a lot on that auction, but lots can still contain a lot from the other one. That is why the code stores lot.domain.name for every lot instead of assuming the auction you asked for. SQLite and PostgreSQL both support this ON CONFLICT … DO UPDATE syntax; see the SQLite upsert documentation↗ for the details of excluded.

The hourly update with /cars?minutes=60

minutes=N returns vehicles whose record or any active lot changed in the last N minutes. It is a rolling lookback measured from the moment the API receives the request, not a cursor. If your job runs at 10:00 and again at 11:03, a fixed minutes=60 misses three minutes of changes. Compute the window from your last successful run instead and add an overlap:

window = max(60, minutes since the last successful run) + 15 normal hourly run, 62 minutes since the last  ->  minutes=77job failed twice, 3 hours 5 minutes since     ->  minutes=200server was down for four days                 ->  over 4320: re-import

The API accepts 1 to 4320 minutes (72 hours); anything larger returns 400. After a longer outage the incremental feed cannot cover the gap, so run a full import and a reconcile instead of trusting it. Polling more often than every 10 to 15 minutes brings almost nothing new, and hourly is what most catalogs need.

Retiring sold lots with /archived-lots

/archived-lots?minutes=W lists lots that left the active inventory in the window: sold, not sold, or removed. Each event carries lot_id, car_id, the VIN, the lot number, the auction, a status and, in the current shape, bid, buy_now, sale_date and final_bid as { value, updated_at } objects. Some older accounts receive a legacy shape where bid is a plain number; the code above detects which one it got and never treats a legacy bid as a confirmed sale price.

Three rules for this step:

1. Archive by lot id. Set archived = 1 on that lot only. If the vehicle has another active lot, it stays in your catalog.
2. Do not delete. An archived lot with a final bid is the history that VIN pages and price research run on. Hide it from the active catalog with a filter, keep the row.
3. Label it honestly. Show "Sold for" only when final_bid exists. An archived lot without one ended without a recorded final price; call it "Ended", not "Sold".

def final_bid_of(event):    # Current shape: "final_bid": {"value": 8450, "updated_at": "..."}.    # The legacy shape has a plain "bid" that is NOT a confirmed sale    # price, so it never fills final_bid.    final = event.get("final_bid")    if isinstance(final, dict):        value = final.get("value")        if isinstance(value, (int, float)) and value > 0:            return value    return None  def save_archive(db, event, now):    if not isinstance(event, dict):        return    lot_id, car_id = event.get("lot_id"), event.get("car_id")    if not is_id(lot_id) or not is_id(car_id):        log.warning("skipped a malformed archive event")        return    # Archive the LOT. The vehicle row is untouched: it may still have    # another active lot, for example on the other auction.    db.execute(        """INSERT INTO lots (id, vehicle_id, domain, lot, archived,                             archived_at, final_bid, archive_payload,                             synced_at)           VALUES (?, ?, ?, ?, 1, ?, ?, ?, ?)           ON CONFLICT(id) DO UPDATE SET archived = 1,             archived_at = excluded.archived_at,             final_bid = COALESCE(excluded.final_bid, lots.final_bid),             archive_payload = excluded.archive_payload,             synced_at = excluded.synced_at""",        (lot_id, car_id, source_of(event), event.get("lot"),         event.get("archived_at"), final_bid_of(event),         json.dumps(event), now))

Run the archive feed after /cars, and widen its window by the time the first feed took. On a first import that pages through the whole inventory, this is what catches lots that sold while the import was still running.

The run function and the checkpoint

The checkpoint is the start time of the last run that completed both feeds. Saving the start time, not the end, means the next window covers everything that changed while this run was busy. Each page is committed in its own transaction, so a crash halfway leaves the finished pages in place; the next run simply replays the window, and the idempotent upserts make the replay harmless.

MAX_WINDOW = 4320  # minutes (72 hours); larger values return 400OVERLAP = 15       # minutes of overlap between runs  def run(path, cars_per_page, archive_per_page, initial):    db = sqlite3.connect(path)    db.executescript(SCHEMA)    started = time.time()    now = datetime.now(timezone.utc).isoformat()    row = db.execute("SELECT finished_at FROM sync_state "                     "WHERE name = 'catalog'").fetchone()    if row is None and not initial:        raise SyncError("no checkpoint yet: run with --initial first")    gap = math.ceil((started - row[0]) / 60) if row else 60    window = max(60, gap) + OVERLAP    if window > MAX_WINDOW and not initial:        raise SyncError("gap over 72 hours: re-import and reconcile")     params = {"per_page": cars_per_page, "simple_paginate": 1}    if not initial:        params["minutes"] = window    for items in pages("cars", params):        with db:  # one transaction per page            for car in items:                save_vehicle(db, car, now)     # Widen the archive window by the time the import itself took.    elapsed = math.ceil((time.time() - started) / 60)    params = {"per_page": archive_per_page, "simple_paginate": 1,              "minutes": min(window + elapsed, MAX_WINDOW)}    for items in pages("archived-lots", params):        with db:            for event in items:                save_archive(db, event, now)     with db:  # the checkpoint moves only after BOTH feeds finished        db.execute("INSERT OR REPLACE INTO sync_state "                   "VALUES ('catalog', ?)", (started,))    db.close()  if __name__ == "__main__":    logging.basicConfig(level=logging.INFO)    run("auction_catalog.sqlite",        int(os.environ.get("CARS_PER_PAGE", "50")),        int(os.environ.get("ARCHIVE_PER_PAGE", "100")),        "--initial" in sys.argv)

Schedule it under a lock so two runs never write at the same time. With cron and flock, a run that is still busy simply makes the next one exit:

# first import, onceAUCTIONS_API_KEY=YOUR_API_KEY CARS_PER_PAGE=1000 \  python sync.py --initial # crontab: every hour, never two at once5 * * * * flock -n /tmp/auction-sync.lock python /srv/sync/sync.py

Errors, retries and backoff

Errors come back as JSON with an error message. Gateways and proxies can still return HTML or an empty body, which the client treats like a network failure. Decide per status what the job does:

The retry budget in get_json is five attempts with delays capped at a minute, which is enough to ride out a deploy or a short network blip. If the budget runs out, the run fails, the checkpoint stays where it was, and the next hourly run covers the gap with a wider window. That is the whole point of computing the window from the checkpoint.

Monitoring a sync job

A sync job fails quietly. The catalog still loads, it is just a little more out of date every hour. Log the endpoint, status, duration, page number and a run id (never the headers or the key), and keep a few numbers per run:

• Age of the last successful run. Alert when it passes two hours. This one metric catches most failures.
• Vehicles upserted and lots archived per run. Zero archived lots on a normal weekday is a sign something is wrong, even if the run reported success.
• Pages and retries per run. A slow rise in retries usually comes before a failure.
• Active lots with a sale date in the past. If they pile up, archive events are being missed and it is time to reconcile.

Reconcile once a week

Rolling windows and overlaps catch almost everything, but not everything. Once a week, page through /cars without minutes and note which lots you saw. Lots that your database holds as active but that did not appear should be checked one by one with /search-lot/{lot}/{domain}; archive them if they come back archived or not found. Do not rely on /archived-lots without minutes for this: it covers roughly the last six months, not the full history.

Test the job without spending requests

Most of what can go wrong here happens in code paths a happy-path run never touches. Save a handful of real responses once, strip anything sensitive, and replay them through the functions above in unit tests instead of calling the API every time. The cases worth a fixture each:

• A response with two pages, where the second page repeats one vehicle from the first (inventory moved while paging).
• A vehicle with one active Copart lot and one active IAAI lot, followed by an archive event for only one of them.
• Both archive shapes, including a legacy event with a plain bid and no final bid.
• Null and missing fields everywhere: no lots, domain set to null, an enum id you have never seen.
• Each error status from the table, plus a 200 with an HTML body, to check that the run stops or retries as intended.
• A run killed after the third page, then started again: the result must match an uninterrupted run.

The last test is the one that proves the job is idempotent, and it is the one teams skip most often.

Before you put it in production

Run the initial import against a copy of your database, then let the hourly job run for a few days and compare counts with a fresh full pass. The sync guide in the documentation has a longer reference script with stricter checks on pagination links, and the /cars and /archived-lots reference pages list every parameter. If you are designing the site around this job, how to build a car auction website covers the database, search index and VIN pages that sit on top of it.