Store Zillow Property Snapshots in Postgres
A Postgres schema and Python importer for property snapshots: current rows, an append-only observation history keyed by collection time, and failed inputs kept for review.
Why one properties table is not enough
The quickest way to store property data is a single table with one row per home, overwritten on every refresh. It answers what the home looks like now and nothing else. You lose the earlier prices you observed, you cannot tell a new collection from a repeated cached one, and inputs that failed simply vanish.
This guide uses four tables instead. A properties table holds the newest observation of each home for fast reads. An observations table keeps every distinct collection, never updated and never deleted by the importer. A failed_inputs table keeps every input that did not return a property. An import_runs table records which snapshot file each row came from, so re-importing a file is harmless.
Choose keys from the response, not from your input
Identify a home by the zpid in the response. An apartment-building link returns no zpid, because a building has none of its own, and the API documents detailUrl as the building's stable identifier; the importer uses it as the fallback key. Never key on the text you sent: two spellings of one address would become two homes.
Identify an observation by the home and meta.fetched_at, the time the source was collected. A cache hit returns the same fetched_at as the earlier live lookup, so the primary key on those two columns turns a repeated observation into a no-op instead of a duplicate row. Store the time as timestamptz so comparisons work across time zones.
| Table | Primary key | Written how |
|---|---|---|
| import_runs | run_id; source_file unique | Once per snapshot file. |
| properties | property_key | Upserted only when the new collection is newer. |
| observations | property_key and fetched_at | Inserted; duplicates ignored. |
| failed_inputs | run_id and item_index | Inserted for each failed item. |
The schema
The body of each observation is stored as jsonb, which Postgres recommends for most JSON use. Keeping the full response means you can add a column later without fetching every home again. A few fields you filter on often, such as status, price and the two estimates, are copied into typed columns. Prices are numeric and nullable: a missing price stays missing, never zero.
CREATE TABLE IF NOT EXISTS import_runs (
run_id bigserial PRIMARY KEY,
source_file text NOT NULL UNIQUE,
requested_at timestamptz NOT NULL,
imported_at timestamptz NOT NULL DEFAULT now()
);
-- One row per home: the newest observation we hold.
CREATE TABLE IF NOT EXISTS properties (
property_key text PRIMARY KEY, -- zpid, or detailUrl for an apartment building
zpid text,
detail_url text,
address text,
first_seen_at timestamptz NOT NULL DEFAULT now(),
last_fetched_at timestamptz NOT NULL,
current jsonb NOT NULL
);
-- Every distinct collection we received. Never updated, never deleted by the importer.
CREATE TABLE IF NOT EXISTS observations (
property_key text NOT NULL REFERENCES properties (property_key),
fetched_at timestamptz NOT NULL, -- meta.fetched_at: when the source was collected
run_id bigint NOT NULL REFERENCES import_runs (run_id),
home_status text,
price numeric,
zestimate numeric,
rent_zestimate numeric,
body jsonb NOT NULL,
PRIMARY KEY (property_key, fetched_at)
);
CREATE TABLE IF NOT EXISTS failed_inputs (
run_id bigint NOT NULL REFERENCES import_runs (run_id),
item_index integer NOT NULL,
input text NOT NULL,
error_code text NOT NULL,
error_message text,
PRIMARY KEY (run_id, item_index)
);Save each batch response before importing it
Separate collection from loading. Write every batch response to a file, together with the inputs you sent and the time you sent them, and import files afterwards. If the database is unavailable you have lost nothing, and you can rebuild the database from the files at any time.
The batch endpoint takes up to 50 links or addresses per request and returns one result per item in the same order, each with its own ok flag. The fetch step below sends one request and saves it; loop over chunks for a longer list.
import os, requests
from import_snapshots import save_snapshot
items = [line.strip() for line in open("links.txt", encoding="utf-8") if line.strip()][:50]
response = requests.post("https://api.propertiesapi.com/v1/zillow/batch",
json={"items": items},
headers={"x-api-key": os.environ["PROPERTIESAPI_KEY"]},
timeout=(10, 330))
response.raise_for_status()
print(save_snapshot(items, response.json()))The importer
Each file is imported in one transaction. The run is registered first; if the file name is already present the importer stops and reports it, which makes a repeated import safe. Each failed item goes to failed_inputs with its error code. Each successful item upserts the properties row only when its collection is newer than the stored one, then inserts the observation and ignores it if that collection is already stored.
The conditional upsert matters when files arrive out of order. Importing an older snapshot after a newer one adds the older observation to the history but leaves the current row alone.
"""Import saved batch snapshots into Postgres.
Usage: python import_snapshots.py snapshots/*.json (DATABASE_URL must be set)
Each snapshot file is {"requested_at", "items", "response"} as written by save_snapshot().
Requires Python 3.10+ and psycopg 3.
"""
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
import psycopg
from psycopg.types.json import Jsonb
def save_snapshot(items, response_body, directory="snapshots"):
"""Call this right after POST /v1/zillow/batch returns 200."""
now = datetime.now(timezone.utc)
path = Path(directory) / f"batch-{now:%Y%m%dT%H%M%S%fZ}.json"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps({"requested_at": now.isoformat(), "items": items,
"response": response_body}), encoding="utf-8")
return path
def import_file(conn, path):
snapshot = json.loads(Path(path).read_text(encoding="utf-8"))
with conn.transaction():
run = conn.execute(
"INSERT INTO import_runs (source_file, requested_at) VALUES (%s, %s) "
"ON CONFLICT (source_file) DO NOTHING RETURNING run_id",
(Path(path).name, snapshot["requested_at"])).fetchone()
if run is None:
return "already imported"
run_id, new, repeat, failed = run[0], 0, 0, 0
for result in snapshot["response"]["data"]:
index = result["index"]
if not result.get("ok"):
error, item = result.get("error") or {}, snapshot["items"][index]
conn.execute("INSERT INTO failed_inputs VALUES (%s, %s, %s, %s, %s)",
(run_id, index, item if isinstance(item, str) else json.dumps(item),
error.get("code", "unknown"), error.get("message")))
failed += 1
continue
data, meta = result["data"], result["meta"]
key = data.get("zpid") or data.get("detailUrl")
conn.execute(
"INSERT INTO properties AS p (property_key, zpid, detail_url, address, "
"last_fetched_at, current) VALUES (%s, %s, %s, %s, %s, %s) "
"ON CONFLICT (property_key) DO UPDATE SET address = EXCLUDED.address, "
"detail_url = EXCLUDED.detail_url, last_fetched_at = EXCLUDED.last_fetched_at, "
"current = EXCLUDED.current WHERE p.last_fetched_at < EXCLUDED.last_fetched_at",
(key, data.get("zpid"), data.get("detailUrl"), data.get("address"),
meta["fetched_at"], Jsonb(data)))
inserted = conn.execute(
"INSERT INTO observations VALUES (%s, %s, %s, %s, %s, %s, %s, %s) "
"ON CONFLICT (property_key, fetched_at) DO NOTHING RETURNING 1",
(key, meta["fetched_at"], run_id, data.get("homeStatus"), data.get("price"),
data.get("zestimate"), data.get("rentZestimate"),
Jsonb({"data": data, "meta": meta}))).fetchone()
new, repeat = (new + 1, repeat) if inserted else (new, repeat + 1)
return f"{new} new observations, {repeat} already stored, {failed} failed inputs"
def main(paths):
with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
conn.execute(Path(__file__).with_name("schema.sql").read_text(encoding="utf-8"))
for path in sorted(paths):
print(f"{path}: {import_file(conn, path)}")
if __name__ == "__main__":
main(sys.argv[1:])Run it and query the history
Install psycopg and point DATABASE_URL at your database. The importer creates the tables if they do not exist. The queries after the import command answer questions that are hard to answer with a single overwritten table: the price path you observed for one home, homes whose newest observation is getting old, and inputs that need attention.
As the history grows, add indexes for the questions you ask most. The primary keys already cover lookups by home and collection time. An index on properties.last_fetched_at helps staleness checks, and a GIN index on the jsonb body helps if you query inside it often. Measure before adding more: each index costs space and slows inserts slightly.
pip install "psycopg[binary]" requests
export DATABASE_URL="postgresql://user@localhost/research"
python import_snapshots.py snapshots/*.json
# snapshots/batch-20261022T070000123456Z.json: 4 new observations, 0 already stored, 1 failed inputs
# Running the same command again prints "already imported" for each file.
-- Price history as you observed it, one row per distinct collection
SELECT fetched_at, home_status, price, zestimate
FROM observations WHERE property_key = '1000001' ORDER BY fetched_at;
-- Homes whose newest observation is older than a week
SELECT property_key, address, last_fetched_at
FROM properties WHERE last_fetched_at < now() - interval '7 days';
-- Inputs that never matched, for manual review
SELECT f.input, f.error_code, r.requested_at
FROM failed_inputs f JOIN import_runs r USING (run_id)
ORDER BY r.requested_at DESC;Describe the history honestly
The observations table is the record of what you collected and when, not a complete history of the home. A price that changed and changed back between two of your refreshes leaves no trace here. For the listing's own events, read the priceHistory array stored in each body, and keep the two ideas separate in reports.
Keep retention and access rules in mind. Photo URLs in the body are references to source images, not copies, and you should store and display property data in line with the data-rights page and your own obligations.
Validation scope: the schema and importer were run against a local Postgres server with synthetic snapshots, covering repeated imports, an out-of-order import, a cached repeat and a failed input. No live properties were collected for this article.
Sources and documentation
Product details checked against the linked documentation on October 22, 2026. Code and sample inputs are illustrative unless stated otherwise.
- PropertiesAPI batch endpoint
- PropertiesAPI property fields and meta.fetched_at
- PostgreSQL INSERT and ON CONFLICT
- PostgreSQL JSON types
- Psycopg JSON adaptation
Written by PropertiesAPI, an independent property data service. We are not affiliated with Zillow. Source data may be incomplete or change over time.