139 lines
4.6 KiB
Python
139 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
sbin/migrate_parent_key.py
|
|
==========================
|
|
Adds parent_key column, changes uniqueness from UNIQUE(name) to
|
|
UNIQUE(parent_key, name), and clears device_type/location for sub-components.
|
|
|
|
SQLite: recreates the device table (required to change constraints).
|
|
MariaDB: uses ALTER TABLE. Check the index name with:
|
|
SHOW INDEX FROM device WHERE Column_name = 'name';
|
|
|
|
Usage:
|
|
python sbin/migrate_parent_key.py # SQLite (batteries.db)
|
|
MARIADB_URL='mysql+...' python sbin/migrate_parent_key.py
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def snapshot(src: Path) -> None:
|
|
dst = src.parent / f"{src.name}.{date.today().isoformat()}.snapshot"
|
|
shutil.copy2(src, dst)
|
|
print(f"Snapshot: {dst.name}")
|
|
|
|
|
|
def migrate_sqlite() -> None:
|
|
import sqlite3
|
|
|
|
db_path = REPO_ROOT / "batteries.db"
|
|
if not db_path.exists():
|
|
print("batteries.db not found — nothing to migrate.")
|
|
return
|
|
|
|
snapshot(db_path)
|
|
conn = sqlite3.connect(str(db_path))
|
|
conn.execute("PRAGMA foreign_keys = OFF")
|
|
|
|
# Check if parent_key already exists (idempotency)
|
|
cols = {row[1] for row in conn.execute("PRAGMA table_info(device)")}
|
|
if "parent_key" in cols:
|
|
print("parent_key already present — running data-only fixup.")
|
|
conn.execute("UPDATE device SET parent_key = COALESCE(parent_id, -1)")
|
|
conn.execute("UPDATE device SET device_type = NULL, location = NULL WHERE parent_id IS NOT NULL")
|
|
conn.commit()
|
|
conn.close()
|
|
print("Done.")
|
|
return
|
|
|
|
conn.execute("ALTER TABLE device RENAME TO device_old")
|
|
|
|
conn.execute("""
|
|
CREATE TABLE device (
|
|
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
|
|
name VARCHAR(100) NOT NULL,
|
|
battery_slots INTEGER NOT NULL DEFAULT 1,
|
|
device_type VARCHAR(50),
|
|
battery_size VARCHAR(20),
|
|
location VARCHAR(100),
|
|
notes TEXT,
|
|
ha_entity_id VARCHAR(100),
|
|
parent_id INTEGER REFERENCES device(id) ON DELETE SET NULL,
|
|
parent_key INTEGER NOT NULL DEFAULT -1,
|
|
CONSTRAINT uq_device_parent_key_name UNIQUE (parent_key, name)
|
|
)
|
|
""")
|
|
|
|
conn.execute("""
|
|
INSERT INTO device
|
|
(id, name, battery_slots, device_type, battery_size, location,
|
|
notes, ha_entity_id, parent_id, parent_key)
|
|
SELECT
|
|
id, name, battery_slots,
|
|
CASE WHEN parent_id IS NOT NULL THEN NULL ELSE device_type END,
|
|
battery_size,
|
|
CASE WHEN parent_id IS NOT NULL THEN NULL ELSE location END,
|
|
notes, ha_entity_id, parent_id,
|
|
COALESCE(parent_id, -1)
|
|
FROM device_old
|
|
""")
|
|
|
|
conn.execute("DROP TABLE device_old")
|
|
conn.execute("PRAGMA foreign_keys = ON")
|
|
conn.commit()
|
|
|
|
count = conn.execute("SELECT COUNT(*) FROM device").fetchone()[0]
|
|
print(f"SQLite migration complete — {count} device(s) migrated.")
|
|
conn.close()
|
|
|
|
|
|
def migrate_mariadb(url: str) -> None:
|
|
from sqlalchemy import create_engine, text
|
|
|
|
engine = create_engine(url)
|
|
with engine.connect() as conn:
|
|
# Add column (skip if already present)
|
|
try:
|
|
conn.execute(text(
|
|
"ALTER TABLE device ADD COLUMN parent_key INT NOT NULL DEFAULT -1"
|
|
))
|
|
except Exception:
|
|
print("parent_key column already exists — skipping ADD COLUMN.")
|
|
|
|
conn.execute(text("UPDATE device SET parent_key = COALESCE(parent_id, -1)"))
|
|
conn.execute(text(
|
|
"UPDATE device SET device_type = NULL, location = NULL WHERE parent_id IS NOT NULL"
|
|
))
|
|
|
|
# Drop old UNIQUE index on name — check actual name first:
|
|
# SHOW INDEX FROM device WHERE Column_name = 'name';
|
|
# Common names: 'name' or 'ix_device_name'
|
|
for idx_name in ("name", "ix_device_name"):
|
|
try:
|
|
conn.execute(text(f"ALTER TABLE device DROP INDEX `{idx_name}`"))
|
|
print(f"Dropped index '{idx_name}'.")
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
conn.execute(text(
|
|
"ALTER TABLE device ADD CONSTRAINT uq_device_parent_key_name "
|
|
"UNIQUE (parent_key, name)"
|
|
))
|
|
conn.commit()
|
|
|
|
print("MariaDB migration complete.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mariadb_url = os.environ.get("MARIADB_URL", "").strip()
|
|
if mariadb_url:
|
|
migrate_mariadb(mariadb_url)
|
|
else:
|
|
migrate_sqlite()
|