#!/usr/bin/env python3 """ Idempotent PocketBase schema setup for Kings & Queens family and OTP support. """ import copy import getpass import json import os import re import urllib.error import urllib.request PB_URL = os.environ.get("KQ_PB_URL", "http://127.0.0.1:8091").rstrip("/") def pb_req(method, path, data=None, token=None, request_client=None): url = f"{PB_URL}/api/{path}" body = json.dumps(data).encode("utf-8") if data is not None else None req = urllib.request.Request(url, data=body, method=method) req.add_header("Content-Type", "application/json") if token: req.add_header("Authorization", token) client = request_client or urllib.request.urlopen try: with client(req) as resp: raw = resp.read().decode("utf-8") return (json.loads(raw) if raw else {}, resp.status) except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8") try: return json.loads(raw), exc.code except json.JSONDecodeError: return {"raw": raw}, exc.code except urllib.error.URLError: return {"error": "network"}, 0 def auth(): email = os.environ.get("KQ_PB_ADMIN_EMAIL") or input("Superuser email: ").strip() password = os.environ.get("KQ_PB_ADMIN_PASSWORD") or getpass.getpass("Superuser password: ") data, status = pb_req( "POST", "collections/_superusers/auth-with-password", {"identity": email, "password": password}, ) if status != 200: raise SystemExit(f"Auth failed ({status})") return data["token"] def relation_field(name, collection_id, required=False, cascade=False, max_select=1): return { "name": name, "type": "relation", "required": required, "collectionId": collection_id, "maxSelect": max_select, "cascadeDelete": cascade, } def _unique_values(values): return list(dict.fromkeys(values)) def merge_field(existing, desired): merged = copy.deepcopy(existing) changed = False for key, value in desired.items(): if key == "values": desired_values = desired.get("values") or [] merged_values = _unique_values([*(merged.get("values") or []), *desired_values]) if merged_values != merged.get("values"): merged["values"] = merged_values changed = True continue if merged.get(key) != value: merged[key] = copy.deepcopy(value) changed = True return merged, changed def merge_fields(existing_fields, desired_fields): merged = [copy.deepcopy(field) for field in existing_fields] by_name = {field["name"]: idx for idx, field in enumerate(merged)} changed = False desired_items = desired_fields.values() if isinstance(desired_fields, dict) else desired_fields for desired in desired_items: current_index = by_name.get(desired["name"]) if current_index is None: merged.append(copy.deepcopy(desired)) by_name[desired["name"]] = len(merged) - 1 changed = True continue current = merged[current_index] updated, field_changed = merge_field(current, desired) if field_changed: merged[current_index] = updated changed = True return merged, changed def merge_indexes(existing_indexes, desired_indexes): merged = list(existing_indexes or []) canonical_existing = { re.sub(r"\s+", "", index).lower(): index for index in merged } changed = False for index in desired_indexes or []: canonical = re.sub(r"\s+", "", index).lower() if canonical not in canonical_existing: merged.append(index) canonical_existing[canonical] = index changed = True return merged, changed def build_schema(): teacher_only = "@request.auth.id != '' && @request.auth.rol = 'teacher'" teacher_or_self_user = "@request.auth.id != '' && (@request.auth.rol = 'teacher' || id = @request.auth.id)" teacher_or_self_profile = "@request.auth.id != '' && (@request.auth.rol = 'teacher' || user_id = @request.auth.id)" teacher_or_self_class = "@request.auth.id != '' && (@request.auth.rol = 'teacher' || student_id = @request.auth.id)" rol_field = { "name": "rol", "type": "select", "required": False, "maxSelect": 1, "values": ["teacher", "student", "guardian"], } phone_field = {"name": "phone", "type": "text", "required": False} otp_enabled_field = {"name": "otp_enabled", "type": "bool", "required": False} is_minor_field = {"name": "is_minor", "type": "bool", "required": False} return { "users_fields": { "rol": rol_field, "phone": phone_field, "otp_enabled": otp_enabled_field, }, "student_fields": { "is_minor": is_minor_field, }, "collections": { "guardian_profiles": { "type": "base", "fields": [ relation_field("user_id", "_pb_users_auth_", True, True), {"name": "phone", "type": "text", "required": True}, {"name": "is_active", "type": "bool", "required": False}, ], "indexes": [ "CREATE UNIQUE INDEX idx_guardian_profiles_phone ON guardian_profiles (phone) WHERE phone != ''", "CREATE UNIQUE INDEX idx_guardian_profiles_user ON guardian_profiles (user_id)", ], "listRule": teacher_only, "viewRule": teacher_only, "createRule": teacher_only, "updateRule": teacher_only, "deleteRule": teacher_only, }, "student_guardians": { "type": "base", "fields": [ relation_field("student_id", "_pb_users_auth_", True, True), relation_field("guardian_id", "_pb_users_auth_", True, True), {"name": "is_primary", "type": "bool", "required": False}, ], "indexes": [ "CREATE UNIQUE INDEX idx_student_guardian ON student_guardians (student_id, guardian_id)", "CREATE INDEX idx_student_guardians_student ON student_guardians (student_id)", "CREATE INDEX idx_student_guardians_guardian ON student_guardians (guardian_id)", ], "listRule": teacher_only, "viewRule": teacher_only, "createRule": teacher_only, "updateRule": teacher_only, "deleteRule": teacher_only, }, "otp_codes": { "type": "base", "fields": [ {"name": "phone", "type": "text", "required": True}, {"name": "code_hash", "type": "text", "required": True}, {"name": "expires_at", "type": "date", "required": True}, {"name": "requested_at", "type": "date", "required": True}, {"name": "request_ip", "type": "text", "required": False}, {"name": "attempts", "type": "number", "required": False}, {"name": "consumed", "type": "bool", "required": False}, {"name": "eligible", "type": "bool", "required": False}, ], "indexes": [ "CREATE INDEX idx_otp_codes_phone_requested ON otp_codes (phone, requested_at)", "CREATE INDEX idx_otp_codes_ip_requested ON otp_codes (request_ip, requested_at)", ], "listRule": None, "viewRule": None, "createRule": None, "updateRule": None, "deleteRule": None, }, "otp_identity_grants": { "type": "base", "fields": [ {"name": "grant_hash", "type": "text", "required": True}, {"name": "phone", "type": "text", "required": True}, {"name": "identity_ids", "type": "json", "required": True}, {"name": "expires_at", "type": "date", "required": True}, {"name": "consumed", "type": "bool", "required": False}, ], "listRule": None, "viewRule": None, "createRule": None, "updateRule": None, "deleteRule": None, }, }, "existing_collections": { "users": { "type": "auth", "fields": [ copy.deepcopy(rol_field), copy.deepcopy(phone_field), copy.deepcopy(otp_enabled_field), ], "indexes": [ "CREATE INDEX idx_users_phone ON users (phone)", "CREATE UNIQUE INDEX idx_users_guardian_phone ON users (phone) WHERE rol = 'guardian' AND phone != ''", ], "listRule": teacher_or_self_user, "viewRule": teacher_or_self_user, "createRule": teacher_only, "updateRule": teacher_only, "deleteRule": teacher_only, }, "student_profiles": { "type": "base", "fields": [ copy.deepcopy(is_minor_field), ], "indexes": [ "CREATE INDEX idx_student_profiles_phone ON student_profiles (phone)", "CREATE UNIQUE INDEX idx_student_profiles_user ON student_profiles (user_id)", ], "listRule": teacher_or_self_profile, "viewRule": teacher_or_self_profile, "createRule": teacher_only, "updateRule": teacher_only, "deleteRule": teacher_only, }, "classes": { "type": "base", "fields": [], "listRule": teacher_or_self_class, "viewRule": teacher_or_self_class, "createRule": teacher_only, "updateRule": teacher_only, "deleteRule": teacher_only, }, }, } def get_collection(name, token, request_client=None): return pb_req("GET", f"collections/{name}", token=token, request_client=request_client) def create_collection(name, definition, token, request_client=None): payload = {"name": name, **definition} data, status = pb_req("POST", "collections", payload, token=token, request_client=request_client) if status not in (200, 201, 204): raise RuntimeError(f"Could not create collection {name} ({status})") print(name) return data def update_collection(name, definition, token, request_client=None): current, status = get_collection(name, token, request_client=request_client) if status != 200: raise RuntimeError(f"Cannot update collection {name} (status {status})") current_fields = current.get("fields") or current.get("schema") or [] desired_fields = definition.get("fields") or [] merged_fields, fields_changed = merge_fields(current_fields, desired_fields) current_indexes = current.get("indexes") or [] desired_indexes = definition.get("indexes") or [] merged_indexes, indexes_changed = merge_indexes(current_indexes, desired_indexes) payload = {} for key in ("type", "listRule", "viewRule", "createRule", "updateRule", "deleteRule"): if key in definition and current.get(key) != definition.get(key): payload[key] = copy.deepcopy(definition.get(key)) if fields_changed: payload["fields"] = merged_fields if indexes_changed: payload["indexes"] = merged_indexes if not payload: return current data, status = pb_req("PATCH", f"collections/{current['id']}", payload, token=token, request_client=request_client) if status not in (200, 204): raise RuntimeError(f"Could not update collection {name} ({status})") print(name) return data def ensure_collection(name, definition, token, create_if_missing=False, request_client=None): current, status = get_collection(name, token, request_client=request_client) if status == 200: return update_collection(name, definition, token, request_client=request_client) if status == 404 and create_if_missing: return create_collection(name, definition, token, request_client=request_client) raise RuntimeError(f"Cannot load collection {name} (status {status})") def main(): token = auth() schema = build_schema() for name in ("guardian_profiles", "student_guardians", "otp_codes", "otp_identity_grants"): ensure_collection(name, schema["collections"][name], token, create_if_missing=True) for name in ("users", "student_profiles", "classes"): ensure_collection(name, schema["existing_collections"][name], token, create_if_missing=False) if __name__ == "__main__": main()