72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
|
"""Shared PocketBase connection setup for the scripts in this directory.
|
||
|
|
|
||
|
|
Reads .env.local from the repo root the same way `npm run mcp` does
|
||
|
|
(`tsx --env-file=.env.local`), so the scripts run without splicing
|
||
|
|
credentials onto the command line. Real environment variables win over
|
||
|
|
the file, so CI or a one-off override still works.
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import urllib.error
|
||
|
|
import urllib.request
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
|
|
|
||
|
|
|
||
|
|
def load_env_local(path: Path = REPO_ROOT / ".env.local") -> None:
|
||
|
|
if not path.exists():
|
||
|
|
return
|
||
|
|
for raw in path.read_text().splitlines():
|
||
|
|
line = raw.strip()
|
||
|
|
if not line or line.startswith("#") or "=" not in line:
|
||
|
|
continue
|
||
|
|
key, _, value = line.partition("=")
|
||
|
|
key = key.strip()
|
||
|
|
value = value.strip().strip('"').strip("'")
|
||
|
|
# Never clobber a variable the caller set explicitly.
|
||
|
|
if key and key not in os.environ:
|
||
|
|
os.environ[key] = value
|
||
|
|
|
||
|
|
|
||
|
|
load_env_local()
|
||
|
|
|
||
|
|
PB = os.environ.get("POCKETBASE_URL") or os.environ.get(
|
||
|
|
"NEXT_PUBLIC_POCKETBASE_URL", "https://pb.mega.infra.apphub.co.za"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def http(method, path, body=None, token=None):
|
||
|
|
data = json.dumps(body).encode() if body is not None else None
|
||
|
|
req = urllib.request.Request(f"{PB}{path}", data=data, method=method)
|
||
|
|
if body is not None:
|
||
|
|
req.add_header("Content-Type", "application/json")
|
||
|
|
if token:
|
||
|
|
req.add_header("Authorization", token)
|
||
|
|
try:
|
||
|
|
with urllib.request.urlopen(req) as r:
|
||
|
|
txt = r.read().decode()
|
||
|
|
return r.status, json.loads(txt) if txt else None
|
||
|
|
except urllib.error.HTTPError as e:
|
||
|
|
return e.code, json.loads(e.read().decode() or "{}")
|
||
|
|
|
||
|
|
|
||
|
|
def superuser_token() -> str:
|
||
|
|
email = os.environ.get("POCKETBASE_ADMIN_EMAIL")
|
||
|
|
password = os.environ.get("POCKETBASE_ADMIN_PASSWORD")
|
||
|
|
if not email or not password:
|
||
|
|
sys.exit(
|
||
|
|
"POCKETBASE_ADMIN_EMAIL / POCKETBASE_ADMIN_PASSWORD not found in "
|
||
|
|
"the environment or .env.local."
|
||
|
|
)
|
||
|
|
code, auth = http(
|
||
|
|
"POST",
|
||
|
|
"/api/collections/_superusers/auth-with-password",
|
||
|
|
{"identity": email, "password": password},
|
||
|
|
)
|
||
|
|
if code != 200:
|
||
|
|
sys.exit(f"AUTH FAILED: {code} {auth}")
|
||
|
|
print(f"auth ok on {PB}")
|
||
|
|
return auth["token"]
|