03 Руководства

Синхронизация каталога

Как держать у себя копию рынка: один раз выгрузить каталог, а дальше забирать только то, что изменилось.

Первая выгрузка

Запрашивайте GET /v1/cars?market=kr&limit=100 и переходите по курсору, пока next_cursor не станет null. Каждая возвращённая машина — одна запись квоты, поэтому полная выгрузка большого рынка требует тарифа с подходящей квотой; на демо — до 20 машин за запрос.

Пагинация курсором

Пагинация только курсором: страница не «плывёт», когда каталог меняется. Параметр offset не поддерживается — запрос с ним получит 400. Поле count в ответе — сколько машин подходит под фильтры, если их не больше 10 000; больше — null.

Опрос ленты изменений

GET /v1/changes?market=kr&since=… отдаёт события с момента since: added — новая машина, disappeared — снята с продажи, price_drop и price_up — цена изменилась на min_pct процентов и больше. Копия рынка обновляется каждый час, поэтому опрашивать чаще смысла нет; насколько она свежа сейчас — в заголовке X-Data-Age-Seconds каждого ответа. Запоминайте момент прошлого опроса и передавайте его в since. Лента есть на тарифах Pro и Max и в демо; каждое событие — одна запись квоты.

Снятые с продажи

Проданная или снятая машина получает is_active=false и остаётся в API ещё 14 дней: её можно запросить по id или выбрать фильтром is_active=false. Ссылки на фото живут около суток — храните id машины, а не ссылку.

Пример на 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)