menu 1 2 3 4 5 7 operationnels

migrations/apply_migrations.py

@@ -0,0 +1,89 @@
+"""
+Application séquentielle des migrations SQL du dossier `migrations/`
+sur la base « bolsa ». À exécuter depuis la racine du projet :
+
+    python migrations/apply_migrations.py
+
+Lit la configuration MySQL depuis les variables d'environnement (.env).
+N'applique que les fichiers non déjà enregistrés dans la table
+`__migrations__` (créée automatiquement).
+"""
+import os
+import sys
+from pathlib import Path
+
+import pymysql
+from dotenv import load_dotenv
+
+load_dotenv()
+
+MIGRATIONS_DIR = Path(__file__).parent
+
+
+def get_connection():
+    return pymysql.connect(
+        host=os.environ.get("MYSQL_HOST", "db"),
+        port=int(os.environ.get("MYSQL_PORT", "3306")),
+        user=os.environ.get("MYSQL_USER", "jfgiraud"),
+        password=os.environ.get("MYSQL_PASSWORD", ""),
+        database=os.environ.get("MYSQL_DATABASE", "bolsa"),
+        charset="utf8mb4",
+        autocommit=False,
+    )
+
+
+def main():
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                "CREATE TABLE IF NOT EXISTS `__migrations__` ("
+                "  `id` int NOT NULL AUTO_INCREMENT,"
+                "  `filename` varchar(255) NOT NULL,"
+                "  `applied_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,"
+                "  PRIMARY KEY (`id`),"
+                "  UNIQUE KEY `idx_filename` (`filename`)"
+                ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4"
+            )
+            conn.commit()
+
+            cur.execute("SELECT filename FROM `__migrations__`")
+            applied = {row[0] for row in cur.fetchall()}
+
+            sql_files = sorted(
+                f for f in MIGRATIONS_DIR.glob("*.sql")
+                if f.name != "schema.sql"
+            )
+
+            if not sql_files:
+                print("Aucune migration à appliquer.")
+                return
+
+            for sql_file in sql_files:
+                if sql_file.name in applied:
+                    print(f"[déjà appliquée] {sql_file.name}")
+                    continue
+
+                print(f"[application] {sql_file.name}")
+                sql = sql_file.read_text(encoding="utf-8")
+                # Découpage sur les ';' pour exécuter instruction par instruction
+                statements = [s.strip() for s in sql.split(";") if s.strip()]
+                for stmt in statements:
+                    cur.execute(stmt)
+                cur.execute(
+                    "INSERT INTO `__migrations__` (filename) VALUES (%s)",
+                    (sql_file.name,),
+                )
+                conn.commit()
+                print(f"[OK] {sql_file.name}")
+
+    except Exception as e:
+        conn.rollback()
+        print(f"[ERREUR] {e}", file=sys.stderr)
+        sys.exit(1)
+    finally:
+        conn.close()
+
+
+if __name__ == "__main__":
+    main()