Mirror and sync Australian property data

How the feed works

Complete Python example

A runnable sync script. Run it daily via cron or a scheduled job to keep your local copy current. The endpoint requires the Starter plan or higher; if you are syncing for a business, commercial use starts at Pro.

import requests
from datetime import datetime, timezone, timedelta

API_KEY = "pr_live_..."
BASE    = "https://api.propradar.com.au/v1"

def sync_changes(since_iso: str) -> str:
    """
    Pull all property changes since `since_iso`, upsert them locally,
    and return the new high-water mark to store for the next run.
    """
    cursor = None
    latest_ts = since_iso

    while True:
        params = {
            "since":      since_iso,
            "event_type": "all",   # "listing_update" | "sold" | "withdrawn" | "all"
            "limit":      20,
        }
        if cursor:
            params["cursor"] = cursor

        r = requests.get(
            f"{BASE}/properties/recently-modified",
            params=params,
            headers={"X-API-Key": API_KEY},
            timeout=30,
        )
        r.raise_for_status()
        data = r.json()

        for event in data["events"]:
            upsert_locally(event)           # your DB logic here
            if event["event_date"] > latest_ts:
                latest_ts = event["event_date"]

        cursor = data["pagination"].get("next_cursor")
        if not cursor:
            break   # all pages consumed

    return latest_ts   # save this as the new since= for next run


def upsert_locally(event: dict) -> None:
    pid        = event["property_id"]
    event_type = event["event_type"]   # "listing_update" | "sold" | "withdrawn"
    summary    = event["summary"]

    # Example: insert/update a local `properties` table
    print(f"[{event_type}] {pid} - {summary.get('suburb')} {summary.get('postcode')}")
    # db.execute("INSERT INTO properties ... ON CONFLICT (property_id) DO UPDATE ...")


if __name__ == "__main__":
    # First run: look back 7 days
    since = (datetime.now(timezone.utc) - timedelta(days=7)).isoformat()
    # Subsequent runs: load from your state store, e.g. a file or DB row
    # since = load_last_run_timestamp()

    new_since = sync_changes(since)
    print(f"Sync complete. Next run: since={new_since}")
    # save_last_run_timestamp(new_since)

Event types