Build your topology graph

{
  "nanoid": "building_...",
  "oldid": 42,
  "name": "Example HQ",
  "timezone": "America/Los_Angeles",
  "panels": [
    {
      "nanoid": "panel_...",
      "oldid": 7,
      "name": "Main switchboard",
      "mains": { "supplyVoltage": 277, "phases": "3ph" },
      "upstreamBreaker": null,
      "dataGateways": [],
      "downstreamPanels": [
        { "nanoid": "panel_...", "oldid": 8, "name": "HVAC panel", "downstreamPanels": [] }
      ]
    }
  ]
}

# Page 1 of buildings
curl -s 'https://api.verdigris.co/v3/buildings?page=1&per-page=1000' \
  -H "Authorization: Bearer $TOKEN"

# A building with its panel tree
curl -s 'https://api.verdigris.co/v3/buildings/building_XXXX' \
  -H "Authorization: Bearer $TOKEN"

# Breakers, page 1, with their circuits
curl -s 'https://api.verdigris.co/v3/breakers?page=1&per-page=1000' \
  -H "Authorization: Bearer $TOKEN"

curl -s -G 'https://api.verdigris.co/v3/panels' \
  -H "Authorization: Bearer $TOKEN" \
  --data-urlencode 'filter={"key":"name","operator":"ilike","type":"string","value":"main"}' \
  --data-urlencode 'per-page=100'

import os
import requests

BASE = "https://api.verdigris.co/v3"
HEADERS = {"Authorization": f"Bearer {os.environ['TOKEN']}"}


def get(path, **params):
    r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=30)
    r.raise_for_status()  # add retry/backoff for 429, see the Rate limits page
    return r.json()


def pages(path, key):
    page = 1
    while True:
        r = requests.get(
            f"{BASE}{path}", headers=HEADERS,
            params={"page": page, "per-page": 1000}, timeout=30,
        )
        body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
        if r.status_code == 404 and body.get("total") == 0:  # an empty list is a 404 with an empty page
            return
        r.raise_for_status()
        yield from body[key]
        if body["currentPage"] >= body["pages"]:
            return
        page += 1


nodes, edges = {}, []


def add_panel(panel, building_id, parent_id=None):
    nodes[panel["nanoid"]] = {"type": "panel", "oldid": panel["oldid"], "name": panel["name"]}
    edges.append((parent_id or building_id, "contains" if parent_id is None else "feeds", panel["nanoid"]))
    for child in panel.get("downstreamPanels") or []:
        add_panel(child, building_id, panel["nanoid"])


for b in pages("/buildings", "buildings"):
    nodes[b["nanoid"]] = {"type": "building", "oldid": b["oldid"], "name": b["name"]}
    detail = get(f"/buildings/{b['nanoid']}")
    for panel in detail["panels"]:
        add_panel(panel, b["nanoid"])

for br in pages("/breakers", "breakers"):
    nodes[br["nanoid"]] = {"type": "breaker", "oldid": br["oldid"], "name": br["name"]}
    if br.get("panel"):
        edges.append((br["panel"]["nanoid"], "contains", br["nanoid"]))
    for c in br["circuits"]:
        nodes[c["nanoid"]] = {"type": "circuit", "oldid": c["oldid"], "phase": c["phase"]}
        edges.append((br["nanoid"], "has", c["nanoid"]))

export {}; // marks this file as a module so top-level await compiles under tsc

const BASE = "https://api.verdigris.co/v3";
const headers = { Authorization: `Bearer ${process.env.TOKEN}` };

type Page<K extends string, T> = { currentPage: number; pages: number; total: number } & Record<K, T[]>;
type Panel = { nanoid: string; oldid: number; name: string; downstreamPanels?: Panel[] };
type Circuit = { nanoid: string; oldid: number; phase: "A" | "B" | "C" };
type Breaker = { nanoid: string; oldid: number; name: string; panel?: { nanoid: string }; circuits: Circuit[] };
type Building = { nanoid: string; oldid: number; name: string; panels: Panel[] };

async function get<T>(path: string, params: Record<string, string | number> = {}): Promise<T | null> {
  const qs = new URLSearchParams(Object.entries(params).map(([k, v]) => [k, String(v)]));
  const res = await fetch(`${BASE}${path}?${qs}`, { headers });
  const text = await res.text();
  const body = text ? JSON.parse(text) : null;
  // A list with no matches is a 404 whose body is an empty page (total 0); any other 404 is a real error.
  if (res.status === 404 && body?.total === 0) return null;
  if (!res.ok) throw new Error(`${res.status} ${text}`); // add retry/backoff for 429
  return body as T;
}

async function* pages<K extends string, T>(path: string, key: K): AsyncGenerator<T> {
  for (let page = 1; ; page++) {
    const body = await get<Page<K, T>>(path, { page, "per-page": 1000 });
    if (!body) return;
    yield* body[key];
    if (body.currentPage >= body.pages) return;
  }
}

const nodes = new Map<string, { type: string; oldid: number; name?: string }>();
const edges: [from: string, rel: string, to: string][] = [];

function addPanel(p: Panel, buildingId: string, parentId?: string) {
  nodes.set(p.nanoid, { type: "panel", oldid: p.oldid, name: p.name });
  edges.push([parentId ?? buildingId, parentId ? "feeds" : "contains", p.nanoid]);
  (p.downstreamPanels ?? []).forEach((c) => addPanel(c, buildingId, p.nanoid));
}

for await (const b of pages<"buildings", Building>("/buildings", "buildings")) {
  nodes.set(b.nanoid, { type: "building", oldid: b.oldid, name: b.name });
  const detail = await get<Building>(`/buildings/${b.nanoid}`);
  detail?.panels.forEach((p) => addPanel(p, b.nanoid));
}

for await (const br of pages<"breakers", Breaker>("/breakers", "breakers")) {
  nodes.set(br.nanoid, { type: "breaker", oldid: br.oldid, name: br.name });
  if (br.panel) edges.push([br.panel.nanoid, "contains", br.nanoid]);
  for (const c of br.circuits) {
    nodes.set(c.nanoid, { type: "circuit", oldid: c.oldid });
    edges.push([br.nanoid, "has", c.nanoid]);
  }
}