Continuity & Exit

import json, os, urllib.request

BASE    = "https://api.supero.dev"
DOMAIN  = "acme-corp"
PROJECT = "550e8400-e29b-41d4-a716-446655440000"
KEY     = os.environ["SUPERO_API_KEY"]        # ak_...

def call(method, path, body=None):
    req = urllib.request.Request(
        BASE + path,
        method=method,
        data=json.dumps(body).encode() if body is not None else None,
        headers={"X-API-Key": KEY, "Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req) as r:
        return json.load(r)

# 1. The data model. Use the PROJECT-scoped schema endpoint: the domain-scoped
#    one returns every schema in the domain, not just this project's.
schemas = call("GET", f"/api/v1/projects/{PROJECT}/schemas")["schemas"]
json.dump(schemas, open("_schemas.json", "w"), indent=2)

# Only object schemas hold records. Qualify with the namespace when there is
# one - a bare name is resolved against the caller's own namespace.
types = []
for s in schemas:
    if s.get("schema_type") != "object":
        continue
    ns = s.get("namespace")
    types.append(f"{ns}:{s['name']}" if ns else s["name"])

# 2. The records. Always send limit - with no pagination parameter at all the
#    response has no pagination object and a paging loop never terminates.
for name in types:
    rows, cursor = [], None
    while True:
        body = {"type": name, "limit": 500}
        if cursor:
            body["cursor"] = cursor
        page = call("POST", f"/api/v1/crud/{DOMAIN}/list-bulk", body)
        rows += page.get("results", [])
        pg = page.get("pagination") or {}
        cursor = pg.get("next_cursor")
        if not pg.get("has_more") or not cursor:
            break

    json.dump(rows, open(f"{name.replace(':', '_')}.json", "w"), indent=2)
    print(f"{name}: {len(rows)} records")

pip install --no-cache-dir --upgrade "supero>=3.5.951"