"""
Version: 1.1.0 (2023-12-20)
Auteur: Kahina et franck - Groupe 2
Commit: bigmoletos@yopmail.com
API IHM pour le projet Qualité de l'Air.
Cette API fournit une interface utilisateur web pour visualiser et prédire
la qualité de l'air à partir des données météorologiques.
Fonctionnalités principales:
- Interface web pour la visualisation des données
- Formulaire de prédiction de la qualité de l'air
- Authentification des utilisateurs
- Historique des prédictions
"""
import os
import sys
from pathlib import Path
import time
import sqlite3
# Ajouter le chemin parent au PYTHONPATH avant les imports
current_dir = Path(__file__).parent
sys.path.append(str(current_dir))
from flask import Flask, render_template, jsonify
from flask_cors import CORS
from flask_login import LoginManager
import logging
from sqlalchemy.exc import OperationalError, SQLAlchemyError
from database.models import db, User
from database.config import config
from database.init_db import init_db
from main.routes import main_bp
from auth.routes import auth_bp
from main.doc_routes import doc_bp
# Configuration du logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
[docs]
def create_app(config_name=None):
"""
Crée et configure l'application Flask.
"""
if config_name is None:
config_name = os.getenv('PYTHON_ENV', 'development')
if config_name == 'test':
config_name = 'testing'
app = Flask(__name__)
app.config.from_object(config[config_name]())
# Configuration selon l'environnement
if os.name == 'nt': # Windows = développement local
app.config['ENV'] = 'development'
app.config['DEBUG'] = True
app.config['TESTING'] = True
app.config['USE_RELOADER'] = True
else: # Linux = production/test
app.config['ENV'] = os.getenv('PYTHON_ENV', 'production')
app.config['DEBUG'] = os.getenv('FLASK_DEBUG', '0') == '1'
app.config['TESTING'] = os.getenv('PYTHON_ENV') == 'test'
app.config['USE_RELOADER'] = False
# Créer les dossiers nécessaires
base_path = Path(__file__).parent
database_path = base_path / 'database'
instance_path = base_path / 'instance'
db_file = database_path / 'db.sqlite'
# Créer les dossiers s'ils n'existent pas
for path in [database_path, instance_path]:
try:
path.mkdir(exist_ok=True)
(path / '.gitkeep').touch(exist_ok=True)
logger.info(f"Dossier créé/vérifié : {path}")
except Exception as e:
logger.error(f"Erreur lors de la création du dossier {path}: {e}")
CORS(app)
# Initialisation des extensions
db.init_app(app)
with app.app_context():
try:
# Fermer toutes les connexions à la base de données
db.session.remove()
db.engine.dispose()
logger.info("Connexions à la base de données fermées")
# Vérifier si la base de données existe et est accessible
try:
User.query.first()
logger.info("Base de données existante et accessible")
except (OperationalError, SQLAlchemyError) as e:
logger.warning(f"Erreur d'accès à la base de données: {e}")
logger.info("Tentative d'initialisation de la base de données")
try:
init_db(app)
logger.info("Base de données initialisée avec succès")
except Exception as init_error:
logger.error(
f"Erreur lors de l'initialisation de la BD: {init_error}"
)
raise
except Exception as e:
logger.error(
f"Erreur lors de la configuration de la base de données: {e}")
raise
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'auth.login'
# Gestion des erreurs
@app.errorhandler(500)
def internal_error(error):
logger.error(f"Erreur 500: {error}")
return jsonify({"error": "Erreur interne du serveur"}), 500
@app.errorhandler(404)
def not_found_error(error):
"""Gestionnaire d'erreur 404 personnalisé"""
logger.error(f"Erreur 404: {error}")
return render_template('error.html',
error_message="Page non trouvée",
error_details=str(error),
suggestions=[
"Vérifiez l'URL saisie",
"Utilisez la navigation du site",
"Retournez à la page d'accueil"
]), 404
@app.errorhandler(OperationalError)
def handle_db_error(error):
"""Gestionnaire d'erreur pour les problèmes de base de données"""
logger.error(f"Erreur de base de données: {error}")
try:
init_db(app)
logger.info("Base de données réinitialisée après erreur")
return jsonify({
"message":
"Base de données réinitialisée, veuillez réessayer"
}), 500
except Exception as e:
logger.error(
f"Échec de la réinitialisation de la base de données: {e}")
return jsonify({"error":
"Erreur critique de base de données"}), 500
@login_manager.user_loader
def load_user(user_id):
try:
return User.query.get(int(user_id))
except (OperationalError, SQLAlchemyError) as e:
logger.error(
f"Erreur lors du chargement de l'utilisateur {user_id}: {e}")
return None
# Enregistrement des blueprints
app.register_blueprint(auth_bp, url_prefix='/auth')
app.register_blueprint(main_bp)
app.register_blueprint(doc_bp)
return app
# Créer l'application pour l'export
app = create_app()
if __name__ == '__main__':
try:
logger.info(f"\nlancement de l'application")
# Désactiver le hot reload en production/test
use_reloader = os.getenv('PYTHON_ENV') == 'development'
app.run(host='0.0.0.0',
port=8093,
debug=os.getenv('FLASK_DEBUG', '0') == '1',
use_reloader=use_reloader)
except Exception as e:
logger.error(f"Erreur lors du démarrage de l'application : {e}")
raise