""" Add the JEI_payment_batches collection and a payment_batch relation on JEI_invoices. Idempotent: if the collection or the relation field already exists, the script reports EXISTS and reuses the existing id. PB v0.39.8 API quirks (verified 2026-09-17, recorded in NOTES.md): 1. Field options for select/relation go at the FIELD ROOT, not inside `options`. 2. POST /api/collections uses key `fields`, not `schema`. 3. GET hides field options for select/relation (data IS stored, just not echoed back). 4. PATCH round-trip "wakes up" newly created collections. If a collection 403s on read with empty listRule, PATCH any rule field (even with the same value) once. Run: python scripts/add_payment_batch_field.py """ import os import sys import json import urllib.request from urllib.error import HTTPError PB_URL = os.environ.get("PB_URL", "https://pb.agent.apphub.co.za") PB_TOKEN = os.environ.get("PB_TOKEN", "") if not PB_TOKEN: env_path = os.path.join(os.path.dirname(__file__), "..", ".env.local") if os.path.exists(env_path): with open(env_path, "r", encoding="utf-8") as fh: for line in fh: if line.startswith("PB_TOKEN="): PB_TOKEN = line.strip().split("=", 1)[1] break if not PB_TOKEN: sys.exit("ERROR: PB_TOKEN not set") HEADERS = {"Authorization": PB_TOKEN, "Content-Type": "application/json"} def pb_get(path): req = urllib.request.Request(PB_URL + path, headers=HEADERS, method="GET") with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read()) def pb_post(path, body): req = urllib.request.Request( PB_URL + path, data=json.dumps(body).encode("utf-8"), headers=HEADERS, method="POST", ) try: with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read()), None except HTTPError as e: return None, json.loads(e.read()) def pb_patch(path, body): req = urllib.request.Request( PB_URL + path, data=json.dumps(body).encode("utf-8"), headers=HEADERS, method="PATCH", ) try: with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read()), None except HTTPError as e: return None, json.loads(e.read()) SUPERUSER_RULES = { "listRule": "", "viewRule": "", "createRule": "", "updateRule": "", "deleteRule": "", } def payment_batches_schema(): return { "name": "JEI_payment_batches", "type": "base", "fields": [ { "name": "batch_ref", "type": "text", "required": True, "min": 1, "max": 50, "pattern": "^[A-Za-z0-9_-]+$", }, { "name": "payment_date", "type": "date", "required": True, "min": "", "max": "", }, { "name": "pop_file", "type": "file", "required": True, "maxSelect": 1, "maxSize": 8388608, "mimeTypes": ["application/pdf", "image/jpeg", "image/png"], "thumbs": [], }, { "name": "notes", "type": "text", "required": False, "min": 0, "max": 1000, "pattern": "", }, { # Optional ZAR amount actually paid for this batch. Defaults to 0 # if not entered (e.g. paid from a £ UK bank account). "name": "zar_paid", "type": "number", "required": False, "min": 0, "max": None, "noDecimal": False, }, ], **SUPERUSER_RULES, } def main(): print(f"PB_URL: {PB_URL}") # 1. Find or create JEI_payment_batches listing = pb_get("/api/collections?perPage=200") batches_id = None for c in listing.get("items", []): if c.get("name") == "JEI_payment_batches": batches_id = c.get("id") break if batches_id: print(f"EXISTS: JEI_payment_batches (id={batches_id})") else: resp, err = pb_post("/api/collections", payment_batches_schema()) if err is not None: sys.exit(f"ERROR: failed to create JEI_payment_batches: {err}") batches_id = resp.get("id") print(f"CREATED: JEI_payment_batches (id={batches_id})") # 2. Find JEI_invoices invoices_id = None for c in listing.get("items", []): if c.get("name") == "JEI_invoices": invoices_id = c.get("id") break if not invoices_id: sys.exit("ERROR: JEI_invoices collection not found") print(f"FOUND: JEI_invoices (id={invoices_id})") # 3. Check whether the relation field already exists inv = pb_get(f"/api/collections/{invoices_id}") field_names = {f.get("name") for f in inv.get("fields", [])} if "payment_batch" in field_names: print("EXISTS: JEI_invoices.payment_batch field") else: new_field = { "name": "payment_batch", "type": "relation", "required": False, "collectionId": batches_id, "cascadeDelete": False, "minSelect": 0, "maxSelect": 1, } # PATCH replaces the full fields array, so send the existing fields # plus the new one appended. patched_fields = list(inv.get("fields", [])) + [new_field] # Strip PB-echoed-only keys that should not be re-sent verbatim. clean_fields = [] for f in patched_fields: clean_fields.append({k: v for k, v in f.items() if k != "options"}) resp, err = pb_patch( f"/api/collections/{invoices_id}", {"fields": clean_fields}, ) if err is not None: sys.exit(f"ERROR: failed to add payment_batch field: {err}") print(f"ADDED: JEI_invoices.payment_batch relation -> {batches_id}") # 3b. Ensure JEI_payment_batches has the zar_paid field (added 2026-09-18). batches = pb_get(f"/api/collections/{batches_id}") batches_field_names = {f.get("name") for f in batches.get("fields", [])} if "zar_paid" in batches_field_names: print("EXISTS: JEI_payment_batches.zar_paid field") else: new_zp = { "name": "zar_paid", "type": "number", "required": False, "min": 0, "max": None, "noDecimal": False, } patched_fields = list(batches.get("fields", [])) + [new_zp] clean_fields = [{k: v for k, v in f.items() if k != "options"} for f in patched_fields] resp, err = pb_patch( f"/api/collections/{batches_id}", {"fields": clean_fields}, ) if err is not None: sys.exit(f"ERROR: failed to add zar_paid field: {err}") print(f"ADDED: JEI_payment_batches.zar_paid number field") # 4. Wake-up round-trip: PATCH each collection's rule fields. Even though # the values are already empty strings, the round-trip activates the # collection's rule machinery in PB v0.39.8. for cid in (batches_id, invoices_id): _, err = pb_patch(f"/api/collections/{cid}", dict(SUPERUSER_RULES)) if err is not None: sys.exit(f"ERROR: wake-up PATCH failed for {cid}: {err}") print(f"PATCHED: {cid} (wake-up round-trip)") print("DONE: schema ready.") if __name__ == "__main__": main()