#!/usr/bin/env python3 """Create the AlistAccess NocoDB base + 7-table schema (idempotent). Verified against NocoDB 2026.08.1 on DSM. Run from any machine with SSH to DSM, or adapt to point NocoDB URL directly. Uses workspace XC-TOKEN. Usage: python create_nocodb_schema.py Env/config at top of file. Idempotent: delete + recreate base each run. """ import paramiko, json, base64 # ---- CONFIG ---- SSH_HOST = "192.168.1.1" SSH_PORT = 9998 SSH_USER = "hoelee" SSH_PASS = "CHANGE_ME_ssh_password" NOCODB_URL = "http://localhost:10380" # reached via SSH on DSM NOCODB_TOKEN = "CHANGE_ME_nocodb_token" # ---------------- def make_ssh(): c = paramiko.SSHClient() c.set_missing_host_key_policy(paramiko.AutoAddPolicy()) c.connect(SSH_HOST, port=SSH_PORT, username=SSH_USER, password=SSH_PASS, timeout=15) return c def main(): ssh = make_ssh() def run(cmd): i, o, e = ssh.exec_command(cmd) return o.read().decode() + e.read().decode() def curl(method, path, body=None): h = f'-H "xc-token: {NOCODB_TOKEN}"' d = "" if body is not None: b64 = base64.b64encode(json.dumps(body).encode()).decode() run(f"echo '{b64}' | base64 -d > /tmp/nc_payload.json") d = "--data-binary @/tmp/nc_payload.json -H 'Content-Type: application/json'" out = run(f'curl -s {h} -X {method} {d} "{NOCODB_URL}{path}"') try: return json.loads(out) except Exception: print("RAW:", out[:600]); return None def jget(r, *keys, d=None): try: for k in keys: r = r[k] return r except Exception: return d # 1. drop existing base if any (find by title) bases = curl("GET", "/api/v2/meta/bases/") for b in jget(bases, "list", d=[]) or []: if b.get("title") == "AlistAccess": curl("DELETE", f"/api/v2/meta/bases/{b['id']}") print("deleted old AlistAccess base", b["id"]) # 2. create base b = curl("POST", "/api/v2/meta/bases/", {"title": "AlistAccess"}) base_id = b["id"] print("BASE:", base_id) def create_table(title, columns): return curl("POST", f"/api/v2/meta/bases/{base_id}/tables", {"title": title, "table_name": title, "type": "table", "columns": columns}) def add_col(tid, title, uidt, **extra): body = {"title": title, "uidt": uidt} body.update(extra) return curl("POST", f"/api/v1/db/meta/tables/{tid}/columns", body) def col_id(tid, title): # v1 GET /columns is removed on NocoDB 2026.08.x — read table meta instead t = curl("GET", f"/api/v2/meta/tables/{tid}") for c in jget(t, "columns", d=[]) or []: if c.get("title") == title: return c.get("id") return None def set_options(tid, title, opts): cid = col_id(tid, title) if not cid: raise RuntimeError(f"column {title} not found on {tid}") return curl("PATCH", f"/api/v2/meta/columns/{cid}", {"colOptions": {"options": [{"title": o, "order": i + 1, "color": None} for i, o in enumerate(opts)]}, "dtxp": ",".join(f"'{o}'" for o in opts)}) def add_ss(tid, title, opts): add_col(tid, title, "SingleSelect") cid = col_id(tid, title) set_options(tid, title, opts) return cid def add_link(child_tid, title, parent_tid, type_="bt"): return add_col(child_tid, title, "LinkToAnotherRecord", parentId=parent_tid, childId=child_tid, type=type_) # ---- tables ---- cust = create_table("Customers", [ {"title": "username", "uidt": "SingleLineText"}, {"title": "password", "uidt": "SingleLineText"}, {"title": "display_name", "uidt": "SingleLineText"}, {"title": "email", "uidt": "Email"}, {"title": "notes", "uidt": "LongText"}, {"title": "alist_user_id", "uidt": "Number"}, {"title": "alist_role_id", "uidt": "Number"}, {"title": "last_sync", "uidt": "DateTime"}, {"title": "last_sync_error", "uidt": "LongText"}, ]); cust = jget(cust, "id") add_ss(cust, "status", ["pending","active","suspended","expired","disabled","deleted"]) add_ss(cust, "sync_status", ["pending","syncing","synced","error","unknown"]) add_col(cust, "sync_required", "Checkbox") prod = create_table("Products", [ {"title": "name", "uidt": "SingleLineText"}, {"title": "description", "uidt": "LongText"}, {"title": "default_duration_days", "uidt": "Number"}, ]); prod = jget(prod, "id") add_ss(prod, "status", ["active","inactive"]) res = create_table("Resources", [ {"title": "name", "uidt": "SingleLineText"}, {"title": "alist_path", "uidt": "SingleLineText"}, {"title": "description", "uidt": "LongText"}, ]); res = jget(res, "id") add_ss(res, "status", ["active","inactive"]) cp = create_table("CustomerProducts", [ {"title": "start_at", "uidt": "Date"}, {"title": "expires_at", "uidt": "Date"}, {"title": "last_sync", "uidt": "DateTime"}, {"title": "last_sync_error", "uidt": "LongText"}, ]); cp = jget(cp, "id") add_ss(cp, "status", ["scheduled","active","expired","revoked","suspended"]) add_ss(cp, "sync_status", ["pending","syncing","synced","error","unknown"]) add_col(cp, "sync_required", "Checkbox") ag = create_table("AccessGrants", [ {"title": "start_at", "uidt": "Date"}, {"title": "expires_at", "uidt": "Date"}, {"title": "revoked_at", "uidt": "DateTime"}, {"title": "revoke_reason", "uidt": "LongText"}, {"title": "last_sync", "uidt": "DateTime"}, {"title": "last_sync_error", "uidt": "LongText"}, ]); ag = jget(ag, "id") add_ss(ag, "permission", ["read","readwrite"]) add_ss(ag, "source", ["product","manual"]) add_ss(ag, "status", ["pending","active","revoked","expired"]) add_ss(ag, "sync_status", ["pending","syncing","synced","error","unknown"]) add_col(ag, "sync_required", "Checkbox") al = create_table("AccessLogs", [ {"title": "username", "uidt": "SingleLineText"}, {"title": "resource_path", "uidt": "SingleLineText"}, {"title": "file_path", "uidt": "SingleLineText"}, {"title": "ip_address", "uidt": "SingleLineText"}, {"title": "user_agent", "uidt": "LongText"}, {"title": "bytes", "uidt": "Number"}, {"title": "timestamp", "uidt": "DateTime"}, ]); al = jget(al, "id") add_ss(al, "action", ["login","view","download","preview","logout"]) add_ss(al, "result", ["success","denied"]) pur = create_table("Purchases", [ {"title": "purchase_date", "uidt": "Date"}, {"title": "amount", "uidt": "Decimal"}, {"title": "payment_method", "uidt": "SingleLineText"}, {"title": "order_id", "uidt": "SingleLineText"}, {"title": "notes", "uidt": "LongText"}, ]); pur = jget(pur, "id") rlog = create_table("ReconcileLogs", [ {"title": "run_id", "uidt": "SingleLineText"}, {"title": "details", "uidt": "LongText"}, {"title": "action", "uidt": "SingleLineText"}, {"title": "timestamp", "uidt": "DateTime"}, ]); rlog = jget(rlog, "id") add_ss(rlog, "drift_type", ["user_missing", "role_missing", "dangling_user_ref", "dangling_role_ref", "user_disabled_drift", "scopes_drift", "base_path_drift", "permission_drift", "role_binding_drift", "username_drift", "verify_failed", "reconcile_error"]) add_ss(rlog, "result", ["repaired", "log_only", "failed", "skipped"]) # ---- links ---- add_link(cp, "customer", cust) add_link(cp, "product", prod) add_link(prod, "Resources", res, type_="mm") # Products <-> Resources M2M add_link(ag, "customer", cust) add_link(ag, "resource", res) add_link(al, "customer", cust) add_link(pur, "customer", cust) add_link(pur, "product", prod) add_link(rlog, "customer", cust) ids = {"base": base_id, "Customers": cust, "Products": prod, "Resources": res, "CustomerProducts": cp, "AccessGrants": ag, "AccessLogs": al, "Purchases": pur, "ReconcileLogs": rlog} print("DONE. IDs:", json.dumps(ids, indent=2)) # persist for later steps open("_nocodb_ids.json", "w").write(json.dumps(ids, indent=2)) ssh.close() if __name__ == "__main__": main()