65 lines
3.0 KiB
Python
65 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Add stable user identity to subscription bindings and access records.
|
|
|
|
The default invocation only reports schema changes. --apply creates a consistent
|
|
SQLite backup before the transaction. Run against an isolated database first.
|
|
"""
|
|
import argparse
|
|
import datetime
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
|
|
|
|
def migrate(db_path: Path, apply: bool = False) -> dict:
|
|
if not db_path.is_file():
|
|
raise ValueError("database must already exist")
|
|
uri = db_path.resolve().as_uri() + ("?mode=rw" if apply else "?mode=ro")
|
|
with sqlite3.connect(uri, uri=True) as db:
|
|
changes = []
|
|
for table in ("sub_bind", "sub_update_record"):
|
|
columns = {row[1] for row in db.execute(f'PRAGMA table_info("{table}")')}
|
|
if not columns:
|
|
raise ValueError(f"required table missing: {table}")
|
|
if "user_id" not in columns:
|
|
changes.append(f"add {table}.user_id")
|
|
duplicates = db.execute(
|
|
"select count(*) from (select username from User group by username having count(*) > 1)"
|
|
).fetchone()[0]
|
|
if duplicates:
|
|
raise ValueError("ambiguous duplicate usernames; resolve identities before migration")
|
|
result = {"apply": apply, "schema_changes": changes}
|
|
if not apply:
|
|
return result
|
|
stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
|
|
backup = db_path.with_name(db_path.name + ".before-user-links." + stamp)
|
|
with sqlite3.connect(backup) as snapshot:
|
|
db.backup(snapshot)
|
|
backup.chmod(0o600)
|
|
db.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
for change in changes:
|
|
table = change.split()[1].split(".")[0]
|
|
db.execute(f'ALTER TABLE "{table}" ADD COLUMN user_id INTEGER REFERENCES User(id)')
|
|
for table in ("sub_bind", "sub_update_record"):
|
|
db.execute(f'UPDATE "{table}" SET user_id=(SELECT id FROM User WHERE username="{table}".user) WHERE user_id IS NULL')
|
|
db.execute("CREATE UNIQUE INDEX IF NOT EXISTS User_username_uindex ON User(username)")
|
|
db.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_sub_bind_user_id ON sub_bind(user_id) WHERE user_id IS NOT NULL")
|
|
db.execute("CREATE INDEX IF NOT EXISTS idx_sub_record_user_id ON sub_update_record(user_id, id)")
|
|
db.commit()
|
|
except BaseException:
|
|
db.rollback()
|
|
raise
|
|
result["backup"] = str(backup)
|
|
result["unresolved_bindings"] = db.execute("select count(*) from sub_bind where user_id is null").fetchone()[0]
|
|
result["unresolved_records"] = db.execute("select count(*) from sub_update_record where user_id is null").fetchone()[0]
|
|
return result
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("database", type=Path)
|
|
parser.add_argument("--apply", action="store_true")
|
|
args = parser.parse_args()
|
|
print(json.dumps(migrate(args.database, args.apply), ensure_ascii=False))
|