backupmanager_ynh/sources/init_db.py

#!/usr/bin/env python3
"""Initialise (ou migre) la base de données SQLite. Appelé par les scripts install et upgrade."""
import os
import sqlite3
import sys

config_path = sys.argv[1] if len(sys.argv) > 1 else None
if config_path:
    os.environ["BACKUPMANAGER_CONFIG"] = config_path

# Lire DB_PATH directement depuis le fichier de config (sans importer app/SQLAlchemy)
# pour pouvoir migrer le schéma AVANT que l'import de app ne tente de requêter la DB.
_cfg = {}
if config_path and os.path.exists(config_path):
    with open(config_path) as _f:
        exec(compile(_f.read(), config_path, "exec"), _cfg)
db_path = _cfg.get("DB_PATH") or os.path.join(
    os.path.dirname(os.path.abspath(__file__)), "backupmanager.db"
)

# Migrations SQLite directes — avant tout import de SQLAlchemy/app
if os.path.exists(db_path):
    _conn = sqlite3.connect(db_path)
    _cur = _conn.execute("PRAGMA table_info(jobs)")
    existing_cols = {row[1] for row in _cur.fetchall()}

    migrations = [
        ("destination_id",    "ALTER TABLE jobs ADD COLUMN destination_id INTEGER REFERENCES destinations(id)"),
        ("remote_instance_id","ALTER TABLE jobs ADD COLUMN remote_instance_id INTEGER REFERENCES remote_instances(id)"),
    ]
    for col, sql in migrations:
        if col not in existing_cols:
            _conn.execute(sql)
            _conn.commit()
            print(f"Migration : colonne {col} ajoutée à jobs.")

    # Table many-to-many job_destinations
    tables = {r[0] for r in _conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
    if "job_destinations" not in tables:
        _conn.execute("""
            CREATE TABLE job_destinations (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                job_id INTEGER NOT NULL REFERENCES jobs(id),
                dest_type TEXT NOT NULL,
                dest_id INTEGER NOT NULL
            )
        """)
        # Migrer les données existantes depuis destination_id / remote_instance_id
        if "destination_id" in existing_cols:
            for job_id, dest_id, inst_id in _conn.execute(
                "SELECT id, destination_id, remote_instance_id FROM jobs"
            ).fetchall():
                if dest_id:
                    _conn.execute(
                        "INSERT INTO job_destinations (job_id, dest_type, dest_id) VALUES (?, 'ssh', ?)",
                        (job_id, dest_id),
                    )
                if inst_id:
                    _conn.execute(
                        "INSERT INTO job_destinations (job_id, dest_type, dest_id) VALUES (?, 'instance', ?)",
                        (job_id, inst_id),
                    )
        _conn.commit()
        print("Migration : table job_destinations créée et données migrées.")

    _conn.close()

# Import de app après les migrations — SQLAlchemy peut désormais requêter la DB
from app import app, db

with app.app_context():
    db.create_all()
    print("Base de données initialisée.")