"""
Module d'initialisation de la base de données.
Ce module gère la création et l'initialisation de la base de données SQLite,
notamment la création des tables et l'ajout des utilisateurs par défaut.
Version: 1.0.1 (2023-12-20)
"""
import os
import logging
from pathlib import Path
from flask import Flask
from sqlalchemy.exc import OperationalError, SQLAlchemyError
from sqlalchemy import inspect
from database.models import db, User
logger = logging.getLogger(__name__)
[docs]
def create_minimal_app():
"""Crée une application Flask minimale pour l'initialisation."""
app = Flask(__name__)
# Utiliser un fichier SQLite persistant au lieu d'une base en mémoire
db_path = Path(__file__).parent / 'db' / 'db.sqlite'
db_path.parent.mkdir(parents=True, exist_ok=True)
app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{db_path}'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
return app
[docs]
def init_db(app=None):
"""
Initialise la base de données complète.
Cette fonction:
1. Crée toutes les tables définies dans models.py
2. Configure l'utilisateur administrateur
3. Configure les utilisateurs par défaut
4. Vérifie l'intégrité de la base
Parameters:
----------
app : Flask
Instance de l'application Flask. Si None, une application minimale est créée.
"""
if app is None:
app = create_minimal_app()
with app.app_context():
try:
# Vérifier si la base de données est accessible
try:
db.engine.connect()
logger.info("Connexion à la base de données établie")
except OperationalError as e:
logger.error(f"Erreur de connexion à la base de données: {e}")
# Vérifier le dossier de la base de données
db_path = Path(app.config['SQLALCHEMY_DATABASE_URI'].replace(
'sqlite:///', ''))
if not db_path.parent.exists():
db_path.parent.mkdir(parents=True, exist_ok=True)
logger.info(
f"Dossier de base de données créé: {db_path.parent}")
# Supprimer toutes les tables existantes
try:
db.drop_all()
logger.info("Tables existantes supprimées")
except Exception as e:
logger.warning(
f"Erreur lors de la suppression des tables: {e}")
# Créer les tables avec les nouvelles contraintes
try:
db.create_all()
logger.info("Tables créées avec succès")
except Exception as e:
logger.error(f"Erreur lors de la création des tables: {e}")
raise
# Initialiser l'admin
try:
init_admin_user()
except Exception as e:
logger.error(
f"Erreur lors de l'initialisation de l'admin: {e}")
raise
# Initialiser les utilisateurs par défaut
try:
init_default_users()
except Exception as e:
logger.error(
f"Erreur lors de l'initialisation des utilisateurs: {e}")
raise
# Vérifier que les tables sont bien créées
try:
inspector = inspect(db.engine)
tables = inspector.get_table_names()
if 'user' not in tables:
raise Exception("La table 'user' n'a pas été créée")
logger.info(f"Tables dans la base de données: {tables}")
except Exception as e:
logger.error(f"Erreur lors de la vérification des tables: {e}")
raise
# Afficher un résumé des utilisateurs enregistrés
try:
users = User.query.all()
logger.info("=== Résumé des utilisateurs enregistrés ===")
logger.info(f"Nombre total d'utilisateurs : {len(users)}")
for user in users:
logger.info(
f"- {user.username} ({user.email}) {'[Admin]' if user.is_admin else ''}"
)
logger.info("=========================================")
except Exception as e:
logger.error(
f"Erreur lors de la vérification des utilisateurs: {e}")
raise
logger.info("Initialisation de la base de données terminée")
except Exception as e:
logger.error(
f"Erreur lors de l'initialisation de la base de données : {e}")
try:
db.session.rollback()
except:
pass
raise
[docs]
def init_admin_user():
"""Configure l'utilisateur administrateur."""
try:
admin = User(username='admin',
email='contact@iaproject.fr',
is_admin=True)
admin.set_password(os.getenv('ADMIN_PASSWORD', '1234'))
db.session.add(admin)
db.session.commit()
logger.info("Utilisateur admin créé")
except Exception as e:
logger.error(f"Erreur lors de la création de l'admin: {e}")
db.session.rollback()
raise
[docs]
def init_default_users():
"""Configure les utilisateurs par défaut."""
users = [{
'username': 'franck',
'email': 'franck@iaproject.fr',
'password': os.getenv('DB_USER1_PASSWORD', '1234'),
'is_admin': False
}, {
'username': 'kahina',
'email': 'kahina@iaproject.fr',
'password': os.getenv('DB_USER2_PASSWORD', '1234'),
'is_admin': False
}, {
'username': 'sylvain',
'email': 'sylvain@iaproject.fr',
'password': os.getenv('DB_USER3_PASSWORD', '1234'),
'is_admin': False
}, {
'username': 'joël',
'email': 'joel@iaproject.fr',
'password': os.getenv('DB_USER4_PASSWORD', '1234'),
'is_admin': False
}]
try:
for user_data in users:
user = User(username=user_data['username'],
email=user_data['email'],
is_admin=user_data['is_admin'])
user.set_password(user_data['password'])
db.session.add(user)
db.session.commit()
logger.info("Utilisateurs par défaut créés")
except Exception as e:
logger.error(
f"Erreur lors de la création des utilisateurs par défaut: {e}")
db.session.rollback()
raise