Real Estate API with Python: Search and Read Listing Data

Choose one source and a small search window

Start with a single platform and a small output limit. For the Zillow adapter, the documented location selector is a contiguous-US bounding box in west, south, east, north order. The example below describes a small area around Austin. It is an example selector, not a claim that a particular listing will be returned.

A source-specific request uses POST /v1/zillow/listings/search. The unified route, POST /v1/listings/search, adds a platforms array, but it does not make different source selectors interchangeable. Learn the single-source contract before combining results from multiple platforms.

Set transaction to sale and limit to 1 for the first request. Do not add bedroom, price, sort, or polygon filters merely because another real estate API supports them. The current adapter rejects unsupported filters. A small request with verified selectors is easier to debug and gives you a clear baseline for later changes.

{
  "location": {
    "kind": "bbox",
    "bbox": [
      -97.82,
      30.2,
      -97.65,
      30.35
    ]
  },
  "transaction": "sale",
  "limit": 1
}

Keep credentials and request identity on the server

Create a live API key in the dashboard and place it in the REALTYCRAWL_API_KEY environment variable. The script reads that value at runtime. Avoid putting the key in source control, browser JavaScript, screenshots, or saved request logs. A test key can exercise authentication but does not execute source operations.

Give the logical search an Idempotency-Key before sending it. In the example, REALTYCRAWL_REQUEST_KEY is another environment variable: assign a unique value for a new collection, then preserve it if you rerun after an uncertain network outcome. A fresh key means a fresh logical request, not a safer retry.

Keep the same API credential and request body during recovery. Do not rotate keys in the middle of an unresolved retry or reuse one logical key for a changed location. The API can report a conflict when the key and request no longer agree. Request identity should be stored with your job record rather than generated anew inside every retry attempt.

Build a client that retains useful error information

The helper below sends JSON, adds authentication only for data calls, and returns both the response body and headers. It uses a finite timeout and refuses automatic redirects so the authorization header is not forwarded to an unexpected destination. No third-party Python package is required.

HTTP failures raise an exception containing the status, API error code, and request ID. Network errors remain visible to the caller. This first version deliberately leaves retry decisions outside the helper so you can distinguish a malformed request from a temporary failure before adding automation.

Use the helper in a file named realtycrawl_example.py, followed by the next two snippets. The returned headers matter: request cost and idempotent replay information should survive alongside the JSON. Logging the complete headers you sent would expose credentials, so log a selected set of response fields instead.

import json
import os
from urllib.error import HTTPError
from urllib.request import Request, HTTPRedirectHandler, build_opener

BASE = "https://realtycrawl.dev"

class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None

client = build_opener(NoRedirect())

def request_json(path, *, body=None, authenticated=False, request_key=None):
    headers = {"Accept": "application/json"}
    if authenticated:
        headers["Authorization"] = "Bearer " + os.environ["REALTYCRAWL_API_KEY"]
        headers["X-Max-Credits"] = "1"
    if request_key:
        headers["Idempotency-Key"] = request_key
    encoded = None
    if body is not None:
        headers["Content-Type"] = "application/json"
        encoded = json.dumps(body).encode("utf-8")
    request = Request(BASE + path, data=encoded, headers=headers)
    try:
        with client.open(request, timeout=120) as response:
            return json.load(response), response.headers
    except HTTPError as exc:
        raw = exc.read()
        try:
            error_body = json.loads(raw)
        except (ValueError, UnicodeDecodeError):
            error_body = {}
        error = error_body.get("error", {})
        request_id = exc.headers.get("X-Request-Id", "unknown")
        raise RuntimeError(
            f"HTTP {exc.code}; code={error.get('code', 'unknown')}; "
            f"request_id={request_id}; retry_after={exc.headers.get('Retry-After')}"
        ) from exc

Check the operation and selectors before searching

The metadata endpoint is public. Filter its response to the intended platform and country, then look for the exact operation you plan to call. An active listing-read capability does not imply that native-ID resolution or every search selector is available.

The example checks searchPlatformListings and the supported filter names required by this request. It stops before a paid call if the contract does not match. That is more useful than issuing the request anyway and treating an unsupported operation as an empty result.

Availability remains a runtime condition. A capability marked active can still encounter a provider outage, an exhausted shared allowance, or an upstream response that fails validation. Keep error handling even after the preflight check. The metadata establishes whether a request is supported, not a guarantee that a particular listing exists.

capabilities, _ = request_json(
    "/v1/meta/capabilities?platform=zillow&country=US"
)
search = next((item for item in capabilities["data"]
               if item["operation"] == "searchPlatformListings"), None)
required = {"location.bbox", "transaction.sale", "limit"}
if (not search or search["status"] != "active"
        or not required.issubset(set(search["supported_filters"]))):
    raise RuntimeError("This source and selector combination is unavailable")

Run the search and save the complete envelope

Send the bounded request only after the capability check passes. Preserve the response envelope instead of extracting prices and discarding everything else. The meta object carries request identity, completeness, warnings, pagination, and source outcomes that explain how the result was collected.

The example creates a local snapshot file named for the request ID. That file contains response data and selected accounting headers, never the API key. In a production application, use access-controlled storage and a retention policy appropriate to the data you collect. The file is an integration aid, not a complete historical database design.

If data is an empty array, record a successful empty observation rather than inventing a placeholder listing. If the request fails, retain the error separately from successful snapshots. These two outcomes must not collapse into the same “no properties” message in a user interface.

from pathlib import Path

body = {
    "location": {"kind": "bbox", "bbox": [-97.82, 30.2, -97.65, 30.35]},
    "transaction": "sale",
    "limit": 1,
}
result, headers = request_json(
    "/v1/zillow/listings/search",
    body=body,
    authenticated=True,
    request_key=os.environ["REALTYCRAWL_REQUEST_KEY"],
)
snapshot = {
    "response": result,
    "credits_used": headers.get("X-Credits-Used"),
    "idempotent_replay": headers.get("X-Idempotent-Replay") == "true",
}
request_id = result["meta"]["request_id"]
Path(f"observation-{request_id}.json").write_text(
    json.dumps(snapshot, indent=2), encoding="utf-8"
)
print("Listings:", len(result["data"]))
print("Completeness:", result["meta"]["completeness"])

Read a returned listing without confusing it with a refresh

Take the listing ID from the successful search response. Do not substitute a portal’s native ID into the normalized listing route or construct a guessed identifier. The stored observation belongs to the account that collected it and is retained for 24 hours.

The next snippet performs a separate authenticated read, so its one-credit ceiling applies independently from the search request. Check the current operation cost before running both examples. X-Max-Credits is a ceiling for one request; it does not cap the sum of all calls in your script.

The returned observation may contain null fields or an empty media collection. Preserve those outcomes. Most importantly, the stored read is not a fresh portal lookup. To acquire newer evidence, use a supported source collection operation, then compare observation timestamps in your own application.

from urllib.parse import quote

if result["data"]:
    listing_id = result["data"][0]["id"]
    listing, read_headers = request_json(
        "/v1/listings/" + quote(listing_id, safe=""),
        authenticated=True,
    )
    print("Observed at:", listing["data"]["observed_at"])
    print("Read credits:", read_headers.get("X-Credits-Used"))

Classify failures before adding retries

Authentication failures require a credential or account fix. Validation failures require a corrected selector or filter. Retrying either unchanged wastes time and can obscure the actual integration problem. A 404 on a saved listing can indicate an expired or inaccessible observation, rather than a removed source listing.

For rate limits, respect Retry-After and coordinate all workers using the account. For uncertain network outcomes, keep the same logical request key while resolving whether the original search completed. A completed replay reports zero new credits in X-Credits-Used even though the body repeats an earlier response.

A stored failed response may also replay. Do not assume that resending its key will launch another crawl indefinitely. Inspect the error code and retryability, then decide whether a genuinely new collection attempt is appropriate within the remaining budget. Keep that decision in the job controller, where attempt limits and accounting are visible.

Inspect the fields your application will actually use

After the request works, inspect a returned record at the field level. The price amount may be present while currency is missing; geometry may be absent; property identity may remain unresolved. A successful envelope does not guarantee that every optional field needed by your product is populated.

The snippet below prints a small selected view without converting unknown values into defaults. It uses the saved search response, so it makes no additional API call. Run this inspection before designing a database with non-null constraints or promising that every listing has coordinates and media.

For comparisons, parse valid amount strings with Decimal and retain price.kind and price.period alongside currency. For identity, keep the source platform and native ID even when a normalized listing ID is available. For freshness, use observed_at rather than the time Python printed the output. These choices let a small tutorial grow into an integration without changing the meaning of its data later.

for item in result["data"]:
    summary = {
        "id": item["id"],
        "source": item["source"]["platform"],
        "native_id": item["native_id"],
        "observed_at": item["observed_at"],
        "price": item["price"],
        "areas": item["property"].get("areas", []),
        "property_id": item["property"].get("id"),
    }
    print(json.dumps(summary, indent=2))

Turn the example into a dependable integration

Before increasing volume, test the script with saved fixtures for a successful result, an empty result, a missing price, and a failed source. Confirm that your storage preserves decimal-string amounts and that the interface distinguishes unknown values from zero. Check that credentials never appear in logs or exports.

Add a job record that stores request input, logical key, attempt count, response request ID, and accumulated charges. Set a total budget and stop condition independently of the per-request ceiling. This is the foundation for a scheduled collector; a loop around the example is not enough.

Once the single-source workflow is understood, evaluate another source using its documented selector and capability response. Keep source-specific collection separate from application-level comparison. That lets you expand coverage without silently changing the meaning of the observations already in your dataset.