Incrementally sync real-estate data with Trassets

Maintain a local copy of your data without fetching every table on each run. The tables are reloaded in full once a day, so there is no row-level delta: the unit of change is the table. Ask /changes which tables were reloaded since your last run, then fetch exactly those tables again, completely.

Do not store next_cursor between runs. After a reload, rows can change, disappear or shift position, so resuming at an old cursor silently keeps stale data.

Find the reloaded tables

Call /changes with the since value from your last run. Use the highest last_sync_at from that run’s response; on the very first run, use a timestamp up to 120 days back. Each category table entry carries category and api_table_name; fetch it from /v1/{category}/{api_table_name}.

curl -s "https://api.trassets.ai/changes?since=2026-07-01T00:00:00Z" \
  -H "X-API-Key: $TRASSETS_API_KEY"

Tables missing from the response have not been reloaded. Skip them.

Fetch each changed table in full

Request the first page without cursor. The rows are in data; next_cursor in the response body points to the next page. Pass it unchanged as the cursor parameter and repeat until next_cursor is null. The cursor only pages through a single run.

curl -s "https://api.trassets.ai/v1/property/properties?limit=1000" \
  -H "X-API-Key: $TRASSETS_API_KEY"

Replace the local table

Once all pages are in, replace your local copy of that table rather than appending, so deleted rows disappear as well. Then store the highest last_sync_at as the since value for the next run.

Full Python example

A complete script that re-fetches every table reloaded since the last run. Requires the requests package.

sync_changed_tables.py

import os
import requests

API_KEY = os.environ["TRASSETS_API_KEY"]
BASE_URL = "https://api.trassets.ai"
HEADERS = {"X-API-Key": API_KEY}


def fetch_table(path: str, page_size: int = 1000) -> list[dict]:
    """Fetch all rows of one table, following next_cursor."""
    rows: list[dict] = []
    params: dict = {"limit": page_size}

    while True:
        resp = requests.get(
            f"{BASE_URL}{path}", headers=HEADERS, params=params, timeout=30
        )
        resp.raise_for_status()
        body = resp.json()
        rows.extend(body["data"])

        if body["next_cursor"] is None:
            return rows
        params["cursor"] = body["next_cursor"]


def table_path(entry: dict) -> str:
    """Category tables live under /v1/; raw tables keep their endpoint_url."""
    if entry.get("category"):
        return f"/v1/{entry['category']}/{entry['api_table_name']}"
    return entry["endpoint_url"]


def sync(since: str) -> tuple[dict[str, list[dict]], str]:
    """Re-fetch every table reloaded after `since`.

    Returns the rows per table and the `since` value for the next run.
    """
    resp = requests.get(
        f"{BASE_URL}/changes", headers=HEADERS, params={"since": since}, timeout=30
    )
    resp.raise_for_status()
    changed = resp.json()

    tables = {t["table_name"]: fetch_table(table_path(t)) for t in changed}
    if not changed:
        return tables, since
    # last_sync_at is UTC without a suffix; the API expects it with "Z"
    return tables, max(t["last_sync_at"] for t in changed) + "Z"


if __name__ == "__main__":
    tables, next_since = sync(since="2026-07-01T00:00:00Z")
    for name, rows in tables.items():
        print(f"{name}: {len(rows)} rows")
    # Replace the local tables and persist next_since for the next run.

Get Started

Core Concepts

Guides

Data Model

Incrementally sync real-estate data with Trassets