Procedure de migration des donnée

database/__init__.py

@@ -1,42 +1,60 @@
-from flask_sqlalchemy import SQLAlchemy
+import logging
-from webapp import webapp
+import json
 import os
 
+from flask_sqlalchemy import SQLAlchemy
+from sqlite3 import Cursor, Connection
+from webapp import webapp
+
+
 basedir = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
 webapp.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{os.path.join(basedir, "instance", "database.db")}'
 db = SQLAlchemy(webapp)
 
-def migrate_game_bundle_if_needed():
+def _tableHaveColumn(table_name:str, column_name:str, cursor:Cursor) -> bool: 
-	"""Migre la table game_bundle si elle a l'ancienne structure (id comme clé primaire)"""
+	cursor.execute(f'PRAGMA table_info({table_name})')
-	try:
+	columns = cursor.fetchall()
-		connection = db.session.connection().connection
+	return any(col[1] == column_name for col in columns)
-		cursor = connection.cursor()
+
-		
+def _tableEmpty(table:str, cursor:Cursor) -> bool:
-		cursor.execute("PRAGMA table_info(game_bundle)")
+	return cursor.execute(f'SELECT COUNT(*) FROM {table}').fetchone()[0] == 0
-		columns = cursor.fetchall()
+
-		
+def _renameTable(old_name:str, new_name:str, cursor:Cursor) : 
-		if columns:
+	cursor.execute(f'ALTER TABLE {old_name} RENAME TO {new_name}')
-			has_id_primary = any(col[1] == 'id' and col[5] == 1 for col in columns)
+
-			
+def _dropTable(table_name:str, cursor:Cursor) : 
-			if has_id_primary:
+	cursor.execute(f'DROP TABLE {table_name}')
-				print("Migration de game_bundle: ancienne structure détectée")
+
-				cursor.execute("DROP TABLE game_bundle")
+def _doPreImportMigration(cursor:Cursor):
-				connection.commit()
+	if _tableHaveColumn('game_bundle', 'id', cursor) :
-				print("Ancienne table game_bundle supprimée")
+		logging.info("Table game_bundle détécté, rennomage en game_bundle_old")
-		
+		_renameTable('game_bundle', 'game_bundle_old', cursor)
-		cursor.close()
+
-	except Exception as e:
+def _doPostImportMigration(cursor:Cursor):
-		print(f"Erreur lors de la vérification de migration: {e}")
+	if _tableEmpty('game_bundle', cursor) :
+		logging.info("remplir game_bundle avec game_bundle_old")
+		bundles = cursor.execute(f'SELECT * FROM game_bundle_old').fetchall()
+		for bundle in bundles : 
+			name = bundle[1]
+			json_data = json.loads(bundle[2])
+			url = json_data['url']
+			logging.info(f'import du bundle {name}, {url}')
+			cursor.execute('INSERT INTO game_bundle(url, name, json) VALUES (?, ?, ?)', (url, name, json.dumps(json_data)))
+		logging.info("suppression de la table temporaire game_bundle_old")
+		_dropTable('game_bundle_old', cursor)
 
 with webapp.app_context():
-	migrate_game_bundle_if_needed()
-	
 	with open('database/schema.sql', 'r') as f:
 		sql = f.read()
-		connection = db.session.connection().connection
+		connection : Connection = db.session.connection().connection
 		try:
-			cursor = connection.cursor()
+			cursor : Cursor = connection.cursor()
+			_doPreImportMigration(cursor)
 			cursor.executescript(sql)
+			_doPostImportMigration(cursor)
+			connection.commit()
 			cursor.close()
+		except Exception as e:
+			logging.error(f"lors de l'import de la bdd : {e}")
 		finally:
 			connection.close()