130 lines
4.6 KiB
Python
130 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
One-shot PocketBase migration applied via:
|
|
|
|
PB_URL=https://pb.mega.infra.apphub.co.za \\
|
|
PB_ADMIN_EMAIL=admin@mega.co.za \\
|
|
PB_ADMIN_PASSWORD=... \\
|
|
python3 scripts/migrations/2026-09-28-fuel-logs-previous-reading.py
|
|
|
|
Adds a `previous_reading` self-relation field to FUEL_LOGS so each fill
|
|
row can point to the previous fill for the same vehicle (ordered by
|
|
date_logged). Powers the "previous reading → last reading → next reading"
|
|
chain the fuel team uses to compute consumption at any point in the
|
|
vehicle's history without re-scanning the Buildsite exports.
|
|
|
|
Why a self-relation and not a derived view?
|
|
- PocketBase has no SQL views exposed via the API; a self-relation is
|
|
the only way to make "the row right before this one for this vehicle"
|
|
queryable in one round-trip from the client.
|
|
- The pre-existing `idx_fuel_logs_vehicle_date` index already covers
|
|
the sort+groupBy the backfill walks, so storage cost is just the
|
|
relation column (one id per row that isn't the oldest in its group).
|
|
|
|
Why optional + cascadeDelete=False?
|
|
- The oldest fill per vehicle has no predecessor; the field stays
|
|
unset rather than fabricating a self-reference. The backfill script
|
|
leaves it null in that case.
|
|
- Cascade-delete on a self-relation would nuke every older row when
|
|
a newer one is deleted. We never want that — every fill is an audit
|
|
record. Cascade stays off.
|
|
|
|
Idempotent: re-running is safe; if `previous_reading` is already in
|
|
the field list, the script exits without a PATCH.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
|
|
COLLECTION = "fuel_logs"
|
|
|
|
|
|
def admin_auth(base_url: str) -> str:
|
|
email = os.environ["PB_ADMIN_EMAIL"]
|
|
password = os.environ["PB_ADMIN_PASSWORD"]
|
|
body = json.dumps({"identity": email, "password": password}).encode()
|
|
req = urllib.request.Request(
|
|
f"{base_url}/api/collections/_superusers/auth-with-password",
|
|
data=body,
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read())["token"]
|
|
|
|
|
|
def get_collection(base_url: str, token: str, name: str):
|
|
req = urllib.request.Request(
|
|
f"{base_url}/api/collections/{name}",
|
|
headers={"Authorization": token},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
if e.code == 404:
|
|
return None
|
|
raise
|
|
|
|
|
|
def patch_collection(base_url: str, token: str, name: str, body: dict):
|
|
req = urllib.request.Request(
|
|
f"{base_url}/api/collections/{name}",
|
|
data=json.dumps(body).encode(),
|
|
headers={
|
|
"Authorization": token,
|
|
"Content-Type": "application/json",
|
|
},
|
|
method="PATCH",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
body_txt = e.read().decode() if e.fp else ""
|
|
raise RuntimeError(f"PATCH {name} failed: {e.code} {body_txt}") from e
|
|
|
|
|
|
def main() -> int:
|
|
base_url = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/")
|
|
token = admin_auth(base_url)
|
|
coll = get_collection(base_url, token, COLLECTION)
|
|
if not coll:
|
|
print(f"ERROR: collection {COLLECTION!r} not found on {base_url}")
|
|
return 1
|
|
|
|
fuel_logs_id = coll["id"]
|
|
existing = {f["name"] for f in coll.get("fields", [])}
|
|
if "previous_reading" in existing:
|
|
print("No-op: 'previous_reading' already present on FUEL_LOGS.")
|
|
else:
|
|
new_field = {
|
|
"name": "previous_reading",
|
|
"type": "relation",
|
|
"required": False,
|
|
"collectionId": fuel_logs_id,
|
|
"cascadeDelete": False,
|
|
"maxSelect": 1,
|
|
}
|
|
merged_fields = coll["fields"] + [new_field]
|
|
print(f"Adding 'previous_reading' (self-relation -> {fuel_logs_id}) to {COLLECTION}.")
|
|
patch_collection(base_url, token, COLLECTION, {"fields": merged_fields})
|
|
|
|
# Final sanity dump for the operator.
|
|
final = get_collection(base_url, token, COLLECTION)
|
|
final_field_names = sorted(f["name"] for f in final.get("fields", []))
|
|
print(f"\nFinal field list on {COLLECTION}: {final_field_names}")
|
|
print("\nMigration complete.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except KeyError as e:
|
|
print(f"ERROR: missing environment variable {e}. "
|
|
"Set PB_ADMIN_EMAIL and PB_ADMIN_PASSWORD.", file=sys.stderr)
|
|
sys.exit(2)
|