feat(fuel): PB schema migration for attention_dismissed_at
Adds the optional date field to VEHICLES via a one-shot idempotent migration, and keeps scripts/pb-bootstrap-schema.py in sync so freshly bootstrapped environments also get it. No data backfill — existing rows have the field set to null, which the UI treats as 'not yet dismissed'.
This commit is contained in:
parent
c5230190d0
commit
fb7d3d51b1
2 changed files with 119 additions and 0 deletions
118
scripts/migrations/2026-08-28-vehicles-attention-dismissed.py
Executable file
118
scripts/migrations/2026-08-28-vehicles-attention-dismissed.py
Executable file
|
|
@ -0,0 +1,118 @@
|
|||
#!/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-08-28-vehicles-attention-dismissed.py
|
||||
|
||||
Adds `attention_dismissed_at` (date, optional) to VEHICLES so the Fuel
|
||||
dashboard's "Attention required" panel can hide a vehicle after the
|
||||
user dismisses it. Auto-undismiss: if a fuel log flagged at
|
||||
date_logged > attention_dismissed_at arrives, the vehicle pops back.
|
||||
|
||||
Idempotent: re-running is safe; existing fields are skipped.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
|
||||
COLLECTION = "vehicles"
|
||||
|
||||
NEW_FIELDS = {
|
||||
"attention_dismissed_at": {
|
||||
"name": "attention_dismissed_at",
|
||||
"type": "date",
|
||||
"required": False,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
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
|
||||
|
||||
existing_fields = coll.get("fields", [])
|
||||
existing_field_names = {f["name"] for f in existing_fields}
|
||||
new_fields = [
|
||||
NEW_FIELDS[name] for name in NEW_FIELDS
|
||||
if name not in existing_field_names
|
||||
]
|
||||
if new_fields:
|
||||
merged_fields = existing_fields + new_fields
|
||||
print(f"Adding {len(new_fields)} new field(s) to {COLLECTION}: "
|
||||
f"{[f['name'] for f in new_fields]}")
|
||||
patch_collection(base_url, token, COLLECTION, {"fields": merged_fields})
|
||||
coll = get_collection(base_url, token, COLLECTION)
|
||||
existing_fields = coll.get("fields", [])
|
||||
else:
|
||||
print(f"No-op: all NEW_FIELDS already present on {COLLECTION}.")
|
||||
|
||||
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}. "
|
||||
f"Set PB_ADMIN_EMAIL and PB_ADMIN_PASSWORD.", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
|
@ -180,6 +180,7 @@ DEFS["vehicles"] = {
|
|||
field("description", "text", required=False),
|
||||
field("standard_work_hours", "text", required=False),
|
||||
field("fuel_limits", "text", required=False),
|
||||
field("attention_dismissed_at", "date", required=False),
|
||||
],
|
||||
"indexes": [],
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue