# Data warehouse sync

> Page through all units and their current values and write one JSON line per unit, ready to load into a data warehouse.

The script pages through all units with [`listBatteries`](https://passport.kabin.sh/docs/en/api-reference/catalog/list-batteries), reads the
current values of each one with [`listBatteryValues`](https://passport.kabin.sh/docs/en/api-reference/values/list-battery-values) and writes
`units.jsonl`: one line per unit with `synced_at`, the `unit` and its `values`. Most warehouses load JSON Lines
directly. It needs a key with the scope `read`.

- `values` holds the current value of every unit-level field that has (or had) a value, with its provenance:
  `revision`, `source`, `review_status`, `recorded_at` and, for condition data, `measured_at`. A cleared field has
  `value: null`.
- Model-level data (nominal voltage, rated capacity and so on) belongs to the model version, not the unit: read it once
  per version with [`listModelVersionValues`](https://passport.kabin.sh/docs/en/api-reference/values/list-model-version-values).
- The script makes one request per unit plus one per 200 units. With 10 000 units that is about 17 minutes at the
  shared limit of 600 requests per minute; it waits when it gets `429`.

```bash title="cURL"
#!/usr/bin/env bash
# warehouse-sync.sh: every unit with its current values as one JSON line (needs curl and jq)
set -euo pipefail
auth="Authorization: Bearer $BATTERYPASS_API_KEY"
synced_at=$(date -u +%Y-%m-%dT%H:%M:%SZ)
cursor=""
: > units.jsonl
while :; do
  page=$(curl -sS --fail --retry 5 -G "$BATTERYPASS_URL/api/v1/batteries" -H "$auth" \
    -d limit=200 ${cursor:+--data-urlencode "cursor=$cursor"})
  jq -c '.items[]' <<<"$page" | while IFS= read -r unit; do
    id=$(jq -r .id <<<"$unit")
    curl -sS --fail --retry 5 -H "$auth" "$BATTERYPASS_URL/api/v1/batteries/$id/values" |
      jq -c --argjson unit "$unit" --arg at "$synced_at" '{synced_at: $at, unit: $unit, values: .items}' >> units.jsonl
  done
  cursor=$(jq -r '.next_cursor // empty' <<<"$page")
  [ -n "$cursor" ] || break
done
wc -l < units.jsonl
```

```ts title="TypeScript"
// warehouse-sync.mts: npx tsx warehouse-sync.mts
import { createWriteStream } from 'node:fs';

const BASE = `${process.env.BATTERYPASS_URL}/api/v1`;
const headers = { Authorization: `Bearer ${process.env.BATTERYPASS_API_KEY}` };
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

async function get(path: string) {
  for (let attempt = 1; ; attempt++) {
    const res = await fetch(BASE + path, { headers });
    if (res.ok) return res.json();
    if (res.status !== 429 || attempt === 5) throw new Error(`${path}: ${res.status} ${(await res.json()).code}`);
    await sleep(Number(res.headers.get('Retry-After') ?? 1) * 1000);
  }
}

const out = createWriteStream('units.jsonl');
const syncedAt = new Date().toISOString();
let cursor: string | null = null;
let count = 0;
do {
  const page = await get(`/batteries?limit=200${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`);
  for (const unit of page.items) {
    const values = await get(`/batteries/${unit.id}/values`);
    out.write(JSON.stringify({ synced_at: syncedAt, unit, values: values.items }) + '\n');
    count++;
  }
  cursor = page.next_cursor;
} while (cursor);
out.end();
console.log(`${count} units written to units.jsonl`);
```

```python title="Python"
# warehouse_sync.py
import json
import os
import time
from datetime import datetime, timezone

import requests

BASE = os.environ["BATTERYPASS_URL"] + "/api/v1"
session = requests.Session()
session.headers["Authorization"] = "Bearer " + os.environ["BATTERYPASS_API_KEY"]


def get(path, **params):
    for attempt in range(1, 6):
        res = session.get(BASE + path, params=params, timeout=30)
        if res.status_code != 429 or attempt == 5:
            res.raise_for_status()
            return res.json()
        time.sleep(int(res.headers.get("Retry-After", "1")))


synced_at = datetime.now(timezone.utc).isoformat()
count, cursor = 0, None
with open("units.jsonl", "w", encoding="utf-8") as out:
    while True:
        page = get("/batteries", limit=200, **({"cursor": cursor} if cursor else {}))
        for unit in page["items"]:
            values = get(f"/batteries/{unit['id']}/values")
            out.write(json.dumps({"synced_at": synced_at, "unit": unit, "values": values["items"]}) + "\n")
            count += 1
        cursor = page["next_cursor"]
        if not cursor:
            break
print(f"{count} units written to units.jsonl")
```

```go title="Go"
// warehouse-sync: go run .
package main

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

var (
	base = os.Getenv("BATTERYPASS_URL") + "/api/v1"
	key  = os.Getenv("BATTERYPASS_API_KEY")
)

// get decodes the JSON answer of a GET request into out and waits out 429 answers.
func get(path string, out any) error {
	for attempt := 1; ; attempt++ {
		req, err := http.NewRequest(http.MethodGet, base+path, nil)
		if err != nil {
			return err
		}
		req.Header.Set("Authorization", "Bearer "+key)
		res, err := http.DefaultClient.Do(req)
		if err != nil {
			return err
		}
		if res.StatusCode == http.StatusOK {
			defer res.Body.Close()
			return json.NewDecoder(res.Body).Decode(out)
		}
		res.Body.Close()
		if res.StatusCode != http.StatusTooManyRequests || attempt == 5 {
			return fmt.Errorf("%s: %s", path, res.Status)
		}
		wait, _ := strconv.Atoi(res.Header.Get("Retry-After"))
		time.Sleep(time.Duration(max(wait, 1)) * time.Second)
	}
}

func main() {
	file, err := os.Create("units.jsonl")
	if err != nil {
		log.Fatal(err)
	}
	defer file.Close()
	out := bufio.NewWriter(file)
	defer out.Flush()
	enc := json.NewEncoder(out) // one JSON document per line

	syncedAt := time.Now().UTC().Format(time.RFC3339)
	count, cursor := 0, ""
	for {
		path := "/batteries?limit=200"
		if cursor != "" {
			path += "&cursor=" + url.QueryEscape(cursor)
		}
		var page struct {
			Items      []map[string]any `json:"items"`
			NextCursor *string          `json:"next_cursor"`
		}
		if err := get(path, &page); err != nil {
			log.Fatal(err)
		}
		for _, unit := range page.Items {
			var values struct {
				Items []map[string]any `json:"items"`
			}
			if err := get(fmt.Sprintf("/batteries/%s/values", unit["id"]), &values); err != nil {
				log.Fatal(err)
			}
			line := map[string]any{"synced_at": syncedAt, "unit": unit, "values": values.Items}
			if err := enc.Encode(line); err != nil {
				log.Fatal(err)
			}
			count++
		}
		if page.NextCursor == nil {
			break
		}
		cursor = *page.NextCursor
	}
	fmt.Printf("%d units written to units.jsonl\n", count)
}
```

```json title="units.jsonl (one line, wrapped here)"
{"synced_at":"2026-10-24T02:00:00Z",
 "unit":{"id":"0192a3f0-5b7c-7d21-8e4f-6a2b9c1d3e5f","serial_number":"SV2027-000184","passport_status":"published","row_version":4,"updated_at":"2026-07-02T10:15:00Z"},
 "values":[{"field_key":"condition.state_of_charge","value":81.5,"unit":"%","revision":9,"source":"api","review_status":"approved","measured_at":"2026-10-24T06:00:00Z","recorded_at":"2026-10-24T06:00:03Z","registry_version":"0.5.0","checks":[]}]}
```

Load the file into a staging table and merge on `unit.id`. To read less on later runs, compare `row_version`,
`passport_status` and the values' `revision` with what you stored, as described in
[Sync units](https://passport.kabin.sh/docs/en/guides/sync-units). For a complete copy of everything your organisation holds, an admin can
also download the full organisation export in the portal.