74 lines
2.4 KiB
Python
74 lines
2.4 KiB
Python
|
"""Copy data from a legacy combined data.db into the split databases.
|
||
|
|
|
||
|
|
The legacy database contained users, sensors and readouts in one file. Run the
|
||
|
|
new migrations first, then import the old rows:
|
||
|
|
|
||
|
|
alembic upgrade head
|
||
|
|
alembic -c alembic_users.ini upgrade head
|
||
|
|
python migrate_legacy_db.py
|
||
|
|
|
||
|
|
The legacy file is only read; nothing is deleted from it.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from sqlalchemy import MetaData, Table, create_engine, inspect, select
|
||
|
|
from sqlalchemy.dialects.sqlite import insert
|
||
|
|
|
||
|
|
from models import Sensor, SensorReadout
|
||
|
|
from user_models import User
|
||
|
|
|
||
|
|
|
||
|
|
def _sqlite_engine(path: Path):
|
||
|
|
if not path.exists():
|
||
|
|
raise SystemExit(f"database not found: {path}")
|
||
|
|
return create_engine(f"sqlite:///{path}")
|
||
|
|
|
||
|
|
|
||
|
|
def _copy_table(source, destination, table) -> int:
|
||
|
|
if table.name not in inspect(source).get_table_names():
|
||
|
|
return 0
|
||
|
|
|
||
|
|
source_table = Table(table.name, MetaData(), autoload_with=source)
|
||
|
|
with source.connect() as conn:
|
||
|
|
rows = [dict(row._mapping) for row in conn.execute(select(source_table))]
|
||
|
|
if not rows:
|
||
|
|
return 0
|
||
|
|
|
||
|
|
with destination.begin() as conn:
|
||
|
|
conn.execute(insert(table).on_conflict_do_nothing(), rows)
|
||
|
|
return len(rows)
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
||
|
|
parser.add_argument("--legacy", default="data.db", type=Path)
|
||
|
|
parser.add_argument("--sensors", default="sensors.db", type=Path)
|
||
|
|
parser.add_argument("--main-app", default="main_app.db", type=Path)
|
||
|
|
args = parser.parse_args()
|
||
|
|
|
||
|
|
legacy = _sqlite_engine(args.legacy)
|
||
|
|
sensors = _sqlite_engine(args.sensors)
|
||
|
|
main_app = _sqlite_engine(args.main_app)
|
||
|
|
|
||
|
|
if not inspect(sensors).has_table(Sensor.__tablename__):
|
||
|
|
raise SystemExit(f"{args.sensors} has no schema; run alembic upgrade head first")
|
||
|
|
if not inspect(main_app).has_table(User.__tablename__):
|
||
|
|
raise SystemExit(
|
||
|
|
f"{args.main_app} has no schema; run alembic -c alembic_users.ini upgrade head first"
|
||
|
|
)
|
||
|
|
|
||
|
|
sensor_count = _copy_table(legacy, sensors, Sensor.__table__)
|
||
|
|
readout_count = _copy_table(legacy, sensors, SensorReadout.__table__)
|
||
|
|
user_count = _copy_table(legacy, main_app, User.__table__)
|
||
|
|
|
||
|
|
print(
|
||
|
|
f"Copied {sensor_count} sensors, {readout_count} readouts and "
|
||
|
|
f"{user_count} users. The legacy database was not modified."
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|