94 lines
3.8 KiB
Python
94 lines
3.8 KiB
Python
from functools import wraps
|
|
from flask import current_app, flash, jsonify, request, render_template, redirect, url_for
|
|
from models import db
|
|
from sqlalchemy.exc import SQLAlchemyError, IntegrityError
|
|
from decimal import Decimal, InvalidOperation
|
|
import logging
|
|
|
|
def handle_db_errors(f):
|
|
"""Décorateur pour gérer les erreurs de base de données"""
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
try:
|
|
result = f(*args, **kwargs)
|
|
# Vérifier que le résultat est une réponse Flask valide
|
|
if result is None:
|
|
current_app.logger.error(f'Fonction {f.__name__} a retourné None')
|
|
return render_template('error.html', error="Une erreur inattendue s'est produite."), 500
|
|
return result
|
|
except IntegrityError as e:
|
|
db.session.rollback()
|
|
current_app.logger.error(f'Erreur d\'intégrité dans {f.__name__}: {e}')
|
|
|
|
error_msg = "Une erreur s'est produite lors de l'enregistrement."
|
|
if "UNIQUE constraint failed" in str(e):
|
|
error_msg = "Cette donnée existe déjà."
|
|
elif "CHECK constraint failed" in str(e):
|
|
error_msg = "Les données saisies ne respectent pas les contraintes."
|
|
|
|
if request.is_json:
|
|
return jsonify({'success': False, 'error': error_msg}), 400
|
|
else:
|
|
flash(error_msg, 'error')
|
|
return render_template('error.html', error=error_msg), 400
|
|
|
|
except SQLAlchemyError as e:
|
|
db.session.rollback()
|
|
current_app.logger.error(f'Erreur SQLAlchemy dans {f.__name__}: {e}')
|
|
|
|
error_msg = "Une erreur de base de données s'est produite."
|
|
if request.is_json:
|
|
return jsonify({'success': False, 'error': error_msg}), 500
|
|
else:
|
|
flash(error_msg, 'error')
|
|
return render_template('error.html', error=error_msg), 500
|
|
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
current_app.logger.error(f'Erreur inattendue dans {f.__name__}: {e}')
|
|
|
|
error_msg = "Une erreur inattendue s'est produite."
|
|
if request.is_json:
|
|
return jsonify({'success': False, 'error': error_msg}), 500
|
|
else:
|
|
flash(error_msg, 'error')
|
|
return render_template('error.html', error=error_msg), 500
|
|
|
|
return decorated_function
|
|
|
|
def safe_int_conversion(value, field_name="valeur"):
|
|
"""Conversion sécurisée en entier"""
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return int(value)
|
|
except (ValueError, TypeError):
|
|
raise ValueError(f"La {field_name} doit être un nombre entier valide.")
|
|
|
|
def safe_decimal_conversion(value, field_name="valeur"):
|
|
"""Conversion sécurisée en décimal"""
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return Decimal(str(value))
|
|
except (ValueError, TypeError, InvalidOperation):
|
|
raise ValueError(f"La {field_name} doit être un nombre décimal valide.")
|
|
|
|
def validate_json_data(data, required_fields):
|
|
"""Valide les données JSON et vérifie les champs requis"""
|
|
if not data:
|
|
raise ValueError("Aucune donnée fournie.")
|
|
|
|
missing_fields = [field for field in required_fields if field not in data or data[field] is None]
|
|
if missing_fields:
|
|
raise ValueError(f"Champs requis manquants: {', '.join(missing_fields)}")
|
|
|
|
return True
|
|
|
|
def log_user_action(action, details=None):
|
|
"""Log les actions utilisateur pour audit"""
|
|
current_app.logger.info(f"Action utilisateur: {action}" + (f" - {details}" if details else ""))
|
|
|
|
class ValidationError(Exception):
|
|
"""Exception personnalisée pour les erreurs de validation"""
|
|
pass |