""" Commande Django pour diagnostiquer et tester la connectivité de la base de données Usage: python manage.py check_database """ from django.core.management.base import BaseCommand, CommandError from django.db import connection, connections from django.conf import settings import time import sys class Command(BaseCommand): help = 'Diagnostique la connectivité et la configuration de la base de données' def add_arguments(self, parser): parser.add_argument( '--database', default='default', help='Nom de la base de données à tester (default: default)' ) parser.add_argument( '--stress-test', action='store_true', help='Effectuer un test de stress avec connexions multiples' ) parser.add_argument( '--monitor', type=int, metavar='SECONDS', help='Surveiller la connexion pendant X secondes' ) def handle(self, *args, **options): database_name = options['database'] self.stdout.write( self.style.SUCCESS(f"=== Diagnostic de la base de données '{database_name}' ===\n") ) # Tests de base self._test_basic_connection(database_name) self._test_database_info(database_name) self._test_connection_settings(database_name) # Tests avancés if options['stress_test']: self._stress_test_connections(database_name) if options['monitor']: self._monitor_connection(database_name, options['monitor']) self._test_mysql_variables(database_name) self._provide_recommendations() def _test_basic_connection(self, database_name): """Test de connexion de base""" self.stdout.write("1. Test de connexion de base...") try: conn = connections[database_name] with conn.cursor() as cursor: cursor.execute("SELECT 1") result = cursor.fetchone() if result[0] == 1: self.stdout.write( self.style.SUCCESS(" ✓ Connexion réussie") ) else: self.stdout.write( self.style.ERROR(" ✗ Connexion échouée - résultat inattendu") ) except Exception as e: self.stdout.write( self.style.ERROR(f" ✗ Erreur de connexion : {e}") ) def _test_database_info(self, database_name): """Informations sur la base de données""" self.stdout.write("\n2. Informations sur la base de données...") try: conn = connections[database_name] with conn.cursor() as cursor: # Version MySQL cursor.execute("SELECT VERSION()") version = cursor.fetchone()[0] self.stdout.write(f" Version MySQL : {version}") # Base de données actuelle cursor.execute("SELECT DATABASE()") db_name = cursor.fetchone()[0] self.stdout.write(f" Base de données : {db_name}") # Utilisateur actuel cursor.execute("SELECT USER()") user = cursor.fetchone()[0] self.stdout.write(f" Utilisateur : {user}") # Heure du serveur cursor.execute("SELECT NOW()") server_time = cursor.fetchone()[0] self.stdout.write(f" Heure du serveur : {server_time}") except Exception as e: self.stdout.write( self.style.ERROR(f" ✗ Erreur lors de la récupération des infos : {e}") ) def _test_connection_settings(self, database_name): """Test des paramètres de connexion""" self.stdout.write("\n3. Paramètres de connexion Django...") try: db_settings = settings.DATABASES[database_name] self.stdout.write(f" Engine : {db_settings['ENGINE']}") self.stdout.write(f" Host : {db_settings['HOST']}") self.stdout.write(f" Port : {db_settings['PORT']}") self.stdout.write(f" Database : {db_settings['NAME']}") self.stdout.write(f" User : {db_settings['USER']}") self.stdout.write(f" CONN_MAX_AGE : {db_settings.get('CONN_MAX_AGE', 'Non défini')}") options = db_settings.get('OPTIONS', {}) if options: self.stdout.write(" Options MySQL :") for key, value in options.items(): self.stdout.write(f" {key} : {value}") except Exception as e: self.stdout.write( self.style.ERROR(f" ✗ Erreur lors de la lecture des paramètres : {e}") ) def _test_mysql_variables(self, database_name): """Test des variables MySQL importantes""" self.stdout.write("\n4. Variables MySQL critiques...") important_variables = [ 'wait_timeout', 'interactive_timeout', 'max_connections', 'max_allowed_packet', 'connect_timeout', 'net_read_timeout', 'net_write_timeout', ] try: conn = connections[database_name] with conn.cursor() as cursor: for var in important_variables: cursor.execute(f"SHOW VARIABLES LIKE '{var}'") result = cursor.fetchone() if result: value = result[1] self.stdout.write(f" {var} : {value}") # Warnings pour les valeurs problématiques if var in ['wait_timeout', 'interactive_timeout']: if int(value) < 28800: # 8 heures self.stdout.write( self.style.WARNING(f" ⚠ {var} est peut-être trop faible") ) else: self.stdout.write(f" {var} : Non trouvé") except Exception as e: self.stdout.write( self.style.ERROR(f" ✗ Erreur lors de la lecture des variables : {e}") ) def _stress_test_connections(self, database_name): """Test de stress avec connexions multiples""" self.stdout.write("\n5. Test de stress des connexions...") max_connections = 10 success_count = 0 for i in range(max_connections): try: conn = connections[database_name] with conn.cursor() as cursor: cursor.execute("SELECT CONNECTION_ID()") conn_id = cursor.fetchone()[0] self.stdout.write(f" Connexion {i + 1} : ID {conn_id}") success_count += 1 # Fermer la connexion conn.close() time.sleep(0.1) except Exception as e: self.stdout.write( self.style.ERROR(f" ✗ Connexion {i + 1} échouée : {e}") ) self.stdout.write(f"\n Résultat : {success_count}/{max_connections} connexions réussies") def _monitor_connection(self, database_name, duration): """Surveiller la connexion pendant une durée donnée""" self.stdout.write(f"\n6. Surveillance de la connexion pendant {duration} secondes...") start_time = time.time() check_interval = 5 # Vérifier toutes les 5 secondes while time.time() - start_time < duration: try: conn = connections[database_name] with conn.cursor() as cursor: cursor.execute("SELECT NOW()") server_time = cursor.fetchone()[0] elapsed = int(time.time() - start_time) self.stdout.write(f" [{elapsed}s] Connexion active - {server_time}") except Exception as e: elapsed = int(time.time() - start_time) self.stdout.write( self.style.ERROR(f" [{elapsed}s] ✗ Erreur de connexion : {e}") ) # Tenter une reconnexion try: conn.close() self.stdout.write(" Tentative de reconnexion...") except: pass time.sleep(check_interval) def _provide_recommendations(self): """Fournir des recommandations basées sur les tests""" self.stdout.write("\n" + "=" * 60) self.stdout.write("RECOMMANDATIONS POUR ÉVITER 'MYSQL SERVER HAS GONE AWAY'") self.stdout.write("=" * 60) recommendations = [ "1. Configuration MySQL :", " - wait_timeout = 28800 (8 heures)", " - interactive_timeout = 28800", " - max_allowed_packet = 16M ou plus", "", "2. Configuration Django :", " - CONN_MAX_AGE = 60 (pas plus de 300)", " - CONN_HEALTH_CHECKS = True", " - Ajouter des timeouts dans OPTIONS", "", "3. Code application :", " - Utiliser des transactions courtes", " - Éviter les requêtes très longues", " - Implémenter un middleware de reconnexion", "", "4. cPanel/Hébergement :", " - Vérifier les limites de connexion", " - Configurer les timeouts MySQL", " - Surveiller les ressources serveur", ] for rec in recommendations: if rec.startswith(("1.", "2.", "3.", "4.")): self.stdout.write(self.style.SUCCESS(rec)) else: self.stdout.write(rec)