# Data-Warehouse-Abgleich

> Gehen Sie alle Einheiten und ihre aktuellen Werte seitenweise durch und schreiben Sie je Einheit eine JSON-Zeile, bereit zum Laden in ein Data Warehouse.

Das Skript geht mit [`listBatteries`](https://passport.kabin.sh/docs/de/api-reference/catalog/list-batteries) alle Einheiten seitenweise
durch, liest die aktuellen Werte jeder Einheit mit
[`listBatteryValues`](https://passport.kabin.sh/docs/de/api-reference/values/list-battery-values) und schreibt `units.jsonl`: je Einheit
eine Zeile mit `synced_at`, der `unit` und ihren `values`. Die meisten Data Warehouses laden JSON Lines direkt. Es
braucht einen Schlüssel mit dem Scope `read`.

- `values` enthält den aktuellen Wert jedes Feldes auf Einheitsebene, das einen Wert hat (oder hatte), mit seiner
  Herkunft: `revision`, `source`, `review_status`, `recorded_at` und bei Zustandsdaten `measured_at`. Ein geleertes
  Feld hat `value: null`.
- Daten auf Modellebene (Nennspannung, Nennkapazität und Ähnliches) gehören zur Modellversion, nicht zur Einheit:
  Lesen Sie sie einmal je Version mit
  [`listModelVersionValues`](https://passport.kabin.sh/docs/de/api-reference/values/list-model-version-values).
- Das Skript stellt eine Anfrage je Einheit und eine je 200 Einheiten. Bei 10 000 Einheiten sind das am gemeinsamen
  Limit von 600 Anfragen pro Minute etwa 17 Minuten; bei `429` wartet es.

```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 (eine Zeile, hier umbrochen)"
{"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":[]}]}
```

Laden Sie die Datei in eine Staging-Tabelle und führen Sie sie über `unit.id` zusammen. Um bei späteren Läufen
weniger zu lesen, vergleichen Sie `row_version`, `passport_status` und die `revision` der Werte mit dem, was Sie
gespeichert haben, wie unter [Einheiten abgleichen](https://passport.kabin.sh/docs/de/guides/sync-units) beschrieben. Eine vollständige
Kopie aller Daten Ihrer Organisation kann ein Admin zusätzlich im Portal als Organisationsexport herunterladen.