03 Guides

Keeping a catalog in sync

How to keep a copy of a market on your side: import the catalog once, then take only what has changed.

The first import

Request GET /v1/cars?market=kr&limit=100 and follow the cursor until next_cursor is null. Every car returned is one record of the quota, so a full import of a large market needs a plan with a fitting quota; the demo returns up to 20 cars per request.

Cursor pagination

Pagination is cursor-only, so a page does not shift while the catalog changes. offset is not supported — a request with it gets 400. The count field is how many cars match the filters when there are at most 10,000; above that it is null.

Polling the change feed

GET /v1/changes?market=kr&since=… returns the events since a moment: added — a new car, disappeared — taken off sale, price_drop and price_up — the price changed by min_pct percent or more. A market copy is refreshed every hour, so polling more often makes no sense; how fresh it is right now is in the X-Data-Age-Seconds header of every answer. Remember the moment of the last poll and pass it as since. The feed comes with the Pro and Max plans and the demo; each event costs one record.

Cars taken off sale

A sold or removed car gets is_active=false and stays in the API for 14 more days: you can request it by id or with the filter is_active=false. Photo links live about a day — keep the car's id, not the link.

An example in Python

sync.py

import os
import time
from datetime import datetime, timezone

import httpx

api = httpx.Client(
    base_url="https://api.apiauto.space/v1",
    headers={"X-API-Key": os.environ["APIAUTO_KEY"]},
    timeout=30,
)

def import_market(market: str) -> None:
    """The first import: every page of the catalog, by the cursor."""
    params = {"market": market, "limit": 100}
    while True:
        page = api.get("/cars", params=params).raise_for_status().json()
        for car in page["items"]:
            save(car)                      # your database
        if not page["next_cursor"]:
            break
        params["cursor"] = page["next_cursor"]

def follow_changes(market: str, since: str) -> str:
    """One poll of the feed; returns the moment to ask from next time."""
    now = datetime.now(timezone.utc).isoformat()
    params = {"market": market, "since": since, "limit": 100}
    while True:
        page = api.get("/changes", params=params).raise_for_status().json()
        for event in page["items"]:
            if event["type"] == "added":
                save(api.get(f"/cars/{event['car_id']}").raise_for_status().json())
            elif event["type"] == "disappeared":
                mark_sold(event["car_id"])
            else:                          # price_drop, price_up
                update_price(event["car_id"], event["new_price"])
        if not page["next_cursor"]:
            return now
        params["cursor"] = page["next_cursor"]

import_market("kr")
since = datetime.now(timezone.utc).isoformat()
while True:
    time.sleep(3600)                       # the copy is refreshed hourly
    since = follow_changes("kr", since)