Recipe

Backfill a warehouse after downtime

Problem

You're standing up a new datastore — a warehouse, a search index, a fresh service — or your consumer was offline long enough that replaying webhooks isn't practical. You want a complete, correct snapshot of every CDM entity, and you want the script to be safe to re-run if it dies halfway.

Code

Page each entity from the start, ordered by created_at:asc, and upsert each record. Re-running is a no-op because you key on the record's identity, not on insert order.

backfill.py / backfill.mjs

import os

import requests

BASE = "https://api.propsocket.io/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['PROPSOCKET_API_KEY']}"}
ENTITIES = ["properties", "units", "residents", "leases"]


def page_all(entity):
    offset, limit = 0, 100  # max page size
    while True:
        resp = requests.get(
            f"{BASE}/{entity}",
            headers=HEADERS,
            params={"limit": limit, "offset": offset, "order-by": "created_at:asc"},
        )
        if resp.status_code == 429:
            # Honor the server's backoff instead of guessing.
            import time
            time.sleep(int(resp.headers.get("Retry-After", "1")))
            continue
        resp.raise_for_status()
        body = resp.json()
        yield from body["results"]
        if not body["meta"]["hasMore"]:
            return
        offset += limit


def backfill():
    for entity in ENTITIES:  # order matters: parents before children
        for record in page_all(entity):
            # Upsert keyed on (x_id, integration_id) — re-running is a no-op.
            upsert(entity, record)


if __name__ == "__main__":
    backfill()

const BASE = 'https://api.propsocket.io/v1';
const HEADERS = { Authorization: `Bearer ${process.env.PROPSOCKET_API_KEY}` };
const ENTITIES = ['properties', 'units', 'residents', 'leases'];

async function* pageAll(entity) {
  let offset = 0;
  const limit = 100; // max page size
  for (;;) {
    const url = new URL(`${BASE}/${entity}`);
    url.searchParams.set('limit', String(limit));
    url.searchParams.set('offset', String(offset));
    url.searchParams.set('order-by', 'created_at:asc');
    const resp = await fetch(url, { headers: HEADERS });
    if (resp.status === 429) {
      const wait = Number(resp.headers.get('Retry-After') ?? '1');
      await new Promise((r) => setTimeout(r, wait * 1000));
      continue;
    }
    if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
    const body = await resp.json();
    yield* body.results;
    if (!body.meta.hasMore) return;
    offset += limit;
  }
}

for (const entity of ENTITIES) {
  // order matters: parents before children
  for await (const record of pageAll(entity)) {
    // Upsert keyed on (x_id, integration_id) — re-running is a no-op.
    await upsert(entity, record);
  }
}

package main

import (
	"encoding/json"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"time"
)

const base = "https://api.propsocket.io/v1"

var entities = []string{"properties", "units", "residents", "leases"}

func pageAll(entity string, yield func(map[string]any)) {
	offset, limit := 0, 100 // max page size
	key := os.Getenv("PROPSOCKET_API_KEY")
	for {
		q := url.Values{}
		q.Set("limit", strconv.Itoa(limit))
		q.Set("offset", strconv.Itoa(offset))
		q.Set("order-by", "created_at:asc")
		req, _ := http.NewRequest("GET", base+"/"+entity+"?"+q.Encode(), nil)
		req.Header.Set("Authorization", "Bearer "+key)
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			panic(err)
		}
		if resp.StatusCode == 429 {
			// Honor the server's backoff instead of guessing.
			wait, _ := strconv.Atoi(resp.Header.Get("Retry-After"))
			resp.Body.Close()
			time.Sleep(time.Duration(wait) * time.Second)
			continue
		}
		if resp.StatusCode != 200 {
			resp.Body.Close()
			panic("HTTP " + strconv.Itoa(resp.StatusCode))
		}
		var body struct {
			Meta    struct{ HasMore bool `json:"hasMore"` } `json:"meta"`
			Results []map[string]any                         `json:"results"`
		}
		json.NewDecoder(resp.Body).Decode(&body)
		resp.Body.Close()
		for _, r := range body.Results {
			yield(r)
		}
		if !body.Meta.HasMore {
			return
		}
		offset += limit
	}
}

func main() {
	for _, entity := range entities { // order matters: parents before children
		pageAll(entity, func(record map[string]any) {
			// Upsert keyed on (x_id, integration_id) — re-running is a no-op.
			upsert(entity, record)
		})
	}
}

require "net/http"
require "json"
require "uri"

BASE = "https://api.propsocket.io/v1"
KEY = ENV.fetch("PROPSOCKET_API_KEY")
ENTITIES = %w[properties units residents leases]

def page_all(entity)
  return enum_for(:page_all, entity) unless block_given?
  offset, limit = 0, 100 # max page size
  loop do
    uri = URI("#{BASE}/#{entity}")
    uri.query = URI.encode_www_form(
      limit: limit, offset: offset, "order-by": "created_at:asc"
    )
    req = Net::HTTP::Get.new(uri)
    req["Authorization"] = "Bearer #{KEY}"
    res = Net::HTTP.start(uri.host, uri.port, use_ssl: true) { |h| h.request(req) }
    if res.code.to_i == 429
      sleep(res["Retry-After"].to_i) # honor the server's backoff
      next
    end
    raise "HTTP #{res.code}" unless res.code.to_i == 200
    body = JSON.parse(res.body)
    body["results"].each { |r| yield r }
    return unless body.dig("meta", "hasMore")
    offset += limit
  end
end

ENTITIES.each do |entity| # order matters: parents before children
  page_all(entity).each do |record|
    # Upsert keyed on (x_id, integration_id) — re-running is a no-op.
    upsert(entity, record)
  end
end

<?php
$base = "https://api.propsocket.io/v1";
$key  = getenv("PROPSOCKET_API_KEY");
$entities = ["properties", "units", "residents", "leases"];

function page_all(string $base, string $key, string $entity): Generator
{
    $offset = 0;
    $limit = 100; // max page size
    while (true) {
        $q = http_build_query([
            "limit" => $limit, "offset" => $offset, "order-by" => "created_at:asc",
        ]);
        $ch = curl_init("$base/$entity?$q");
        curl_setopt_array($ch, [
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_HEADER => true,
            CURLOPT_HTTPHEADER => ["Authorization: Bearer $key"],
        ]);
        $raw = curl_exec($ch);
        $status = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
        $headerSize = curl_getinfo($ch, CURLINFO_HEADER_SIZE);
        curl_close($ch);
        if ($status === 429) {
            preg_match("/retry-after:\s*(\d+)/i", substr($raw, 0, $headerSize), $m);
            sleep((int) ($m[1] ?? 1)); // honor the server's backoff
            continue;
        }
        if ($status !== 200) {
            throw new RuntimeException("HTTP $status");
        }
        $body = json_decode(substr($raw, $headerSize), true);
        yield from $body["results"];
        if (!$body["meta"]["hasMore"]) {
            return;
        }
        $offset += $limit;
    }
}

foreach ($entities as $entity) { // order matters: parents before children
    foreach (page_all($base, $key, $entity) as $record) {
        // Upsert keyed on (x_id, integration_id) — re-running is a no-op.
        upsert($entity, $record);
    }
}

import java.net.URI;
import java.net.http.*;
import com.fasterxml.jackson.databind.*; // Jackson: the one JSON dependency

var base = "https://api.propsocket.io/v1";
var key = System.getenv("PROPSOCKET_API_KEY");
var entities = new String[] {"properties", "units", "residents", "leases"};
var client = HttpClient.newHttpClient();
var mapper = new ObjectMapper();

for (String entity : entities) { // order matters: parents before children
    int offset = 0, limit = 100; // max page size
    boolean more = true;
    while (more) {
        var uri = URI.create(base + "/" + entity + "?limit=" + limit
                + "&offset=" + offset + "&order-by=created_at:asc");
        var request = HttpRequest.newBuilder(uri)
                .header("Authorization", "Bearer " + key).build();
        var resp = client.send(request, HttpResponse.BodyHandlers.ofString());
        if (resp.statusCode() == 429) {
            int wait = resp.headers().firstValue("Retry-After")
                    .map(Integer::parseInt).orElse(1);
            Thread.sleep(wait * 1000L); // honor the server's backoff
            continue;
        }
        JsonNode body = mapper.readTree(resp.body());
        for (JsonNode record : body.get("results")) {
            // Upsert keyed on (x_id, integration_id) — re-running is a no-op.
            upsert(entity, record);
        }
        more = body.get("meta").get("hasMore").asBoolean();
        offset += limit;
    }
}

using System.Text.Json;

var BASE = "https://api.propsocket.io/v1";
var key = Environment.GetEnvironmentVariable("PROPSOCKET_API_KEY");
string[] entities = { "properties", "units", "residents", "leases" };
using var client = new HttpClient();
client.DefaultRequestHeaders.Authorization =
    new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", key);

foreach (var entity in entities) // order matters: parents before children
{
    int offset = 0, limit = 100; // max page size
    bool more = true;
    while (more)
    {
        var url = $"{BASE}/{entity}?limit={limit}&offset={offset}&order-by=created_at:asc";
        var resp = await client.GetAsync(url);
        if ((int)resp.StatusCode == 429)
        {
            var wait = resp.Headers.RetryAfter?.Delta?.TotalSeconds ?? 1;
            await Task.Delay(TimeSpan.FromSeconds(wait)); // honor the server's backoff
            continue;
        }
        resp.EnsureSuccessStatusCode();
        using var doc = JsonDocument.Parse(await resp.Content.ReadAsStringAsync());
        var root = doc.RootElement;
        foreach (var record in root.GetProperty("results").EnumerateArray())
        {
            // Upsert keyed on (x_id, integration_id) — re-running is a no-op.
            Upsert(entity, record);
        }
        more = root.GetProperty("meta").GetProperty("hasMore").GetBoolean();
        offset += limit;
    }
}

Why this works

Stable ordering. Sorting by created_at:asc means new records land at the end of the sequence. A record you've already paged past never shifts back into an earlier page, so you don't skip or double-read as the sync engine writes during your run. The contract is in Pagination.
Idempotent upserts. Each CDM record's natural key is itsx_id within an integration (PropSocket enforces a uniqueness constraint on(x_id, integration_id)). The stable id is the simplest upsert key. Either way, the second run overwrites with identical data — no duplicates.
Parents before children. Loading properties, then units, then residents, then leases means foreign-key targets exist before the rows that reference them — no deferred-FK gymnastics on your side.
Honoring Retry-After. A full backfill is the most likely thing to hit the 150 req/min per-Organization limit. Sleeping for the server-suppliedRetry-After keeps you correct without manual tuning.

What to watch out for

This snapshot excludes soft-deletes. By default you get only live records. If you're reconciling — and need to learn what was removed while you were down — add?include_deleted=true and read each record's deleted_at. That's the job of Reconcile after an outage.
Don't parallelize with pre-computed offsets. Firing many pages at once with guessed offsets races against inserts and corrupts the window. Page sequentially; the 100-row max page size keeps even six-figure datasets to a manageable request count.
Leases nest their residents. A lease response includes itsresidents as a relation array (LeaseResident is relation-only, never a top-level collection). You can populate your join table straight from the lease payload rather than cross-referencing separately.
Money stays integer. Persist amount as the integer minor-unit value and keep currency alongside it. Converting to a float during backfill is how rounding bugs get baked into a warehouse permanently.

Once the snapshot is loaded, keep it fresh with webhooks (Listen for new leases; webhooks are on theScale plan and above) and close any gap from your downtime window with Reconcile after an outage.

Backfill a warehouse after downtime

Problem

Code

Why this works

What to watch out for

Next