136 lines
4.5 KiB
Python
136 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Idempotent schema migrations for the radadmin API.
|
|
|
|
Applies incremental DDL that ``radadmin_schema.sql`` can't (it only CREATEs
|
|
tables IF NOT EXISTS, so it never alters an existing one). Each step checks
|
|
information_schema first, so this is safe to re-run and works on both MySQL and
|
|
MariaDB. Uses the same DB connection settings as the app (.env / config.py).
|
|
|
|
Run from the backend/ directory:
|
|
|
|
python migrate.py # apply pending migrations
|
|
python migrate.py --dry-run # show what would run, change nothing
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.database import engine
|
|
|
|
TABLE = "radadmin_clients"
|
|
|
|
|
|
def _column_exists(conn, table: str, column: str) -> bool:
|
|
row = conn.execute(
|
|
text(
|
|
"SELECT COUNT(*) FROM information_schema.columns "
|
|
"WHERE table_schema = DATABASE() AND table_name = :t AND column_name = :c"
|
|
),
|
|
{"t": table, "c": column},
|
|
).scalar()
|
|
return bool(row)
|
|
|
|
|
|
def _index_exists(conn, table: str, index: str) -> bool:
|
|
row = conn.execute(
|
|
text(
|
|
"SELECT COUNT(*) FROM information_schema.statistics "
|
|
"WHERE table_schema = DATABASE() AND table_name = :t AND index_name = :i"
|
|
),
|
|
{"t": table, "i": index},
|
|
).scalar()
|
|
return bool(row)
|
|
|
|
|
|
def _table_exists(conn, table: str) -> bool:
|
|
row = conn.execute(
|
|
text(
|
|
"SELECT COUNT(*) FROM information_schema.tables "
|
|
"WHERE table_schema = DATABASE() AND table_name = :t"
|
|
),
|
|
{"t": table},
|
|
).scalar()
|
|
return bool(row)
|
|
|
|
|
|
# Each migration: (description, predicate that returns True when it still needs
|
|
# to run, SQL to run). Ordered — later steps may depend on earlier ones.
|
|
MIGRATIONS = [
|
|
(
|
|
"add radadmin_clients.groupname (mirror of radusergroup)",
|
|
lambda c: not _column_exists(c, TABLE, "groupname"),
|
|
f"ALTER TABLE {TABLE} ADD COLUMN groupname VARCHAR(64) NULL AFTER alias",
|
|
),
|
|
(
|
|
"add radadmin_clients.status (mirror of customers.status)",
|
|
lambda c: not _column_exists(c, TABLE, "status"),
|
|
f"ALTER TABLE {TABLE} ADD COLUMN status VARCHAR(10) NULL AFTER groupname",
|
|
),
|
|
(
|
|
"add radadmin_clients.deleted_at (soft-delete state)",
|
|
lambda c: not _column_exists(c, TABLE, "deleted_at"),
|
|
f"ALTER TABLE {TABLE} ADD COLUMN deleted_at DATETIME NULL DEFAULT NULL AFTER created_at",
|
|
),
|
|
(
|
|
"add radadmin_clients.active_mac (generated; NULL when soft-deleted)",
|
|
lambda c: not _column_exists(c, TABLE, "active_mac"),
|
|
f"ALTER TABLE {TABLE} ADD COLUMN active_mac VARCHAR(17) "
|
|
"AS (IF(deleted_at IS NULL, mac_address, NULL)) STORED",
|
|
),
|
|
(
|
|
"drop old unique index uq_radadmin_clients_mac",
|
|
lambda c: _index_exists(c, TABLE, "uq_radadmin_clients_mac"),
|
|
f"ALTER TABLE {TABLE} DROP INDEX uq_radadmin_clients_mac",
|
|
),
|
|
(
|
|
"add partial-unique index uq_radadmin_clients_active_mac",
|
|
lambda c: not _index_exists(c, TABLE, "uq_radadmin_clients_active_mac"),
|
|
f"ALTER TABLE {TABLE} ADD UNIQUE KEY uq_radadmin_clients_active_mac (active_mac)",
|
|
),
|
|
]
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Apply radadmin schema migrations.")
|
|
parser.add_argument(
|
|
"--dry-run", action="store_true", help="show pending migrations without applying them"
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
with engine.connect() as conn:
|
|
if not _table_exists(conn, TABLE):
|
|
print(
|
|
f"error: table '{TABLE}' not found — import radadmin_schema.sql first.",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
applied = 0
|
|
for desc, needs_run, sql in MIGRATIONS:
|
|
if not needs_run(conn):
|
|
print(f" skip {desc} (already applied)")
|
|
continue
|
|
if args.dry_run:
|
|
print(f" PEND {desc}\n {sql}")
|
|
applied += 1
|
|
continue
|
|
print(f" apply {desc}")
|
|
conn.execute(text(sql))
|
|
conn.commit() # DDL auto-commits, but be explicit for the connection
|
|
applied += 1
|
|
|
|
if args.dry_run:
|
|
print(f"\n{applied} migration(s) pending. Re-run without --dry-run to apply.")
|
|
elif applied:
|
|
print(f"\nDone — {applied} migration(s) applied.")
|
|
else:
|
|
print("\nDatabase already up to date.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|