Establish a MySQL connection using SQLAlchemy.
(self, ip, username, password, database=None)
| 32 | logger.error(f"Error during initialization: {e}") |
| 33 | |
| 34 | def connect_sql(self, ip, username, password, database=None): |
| 35 | """ |
| 36 | Establish a MySQL connection using SQLAlchemy. |
| 37 | """ |
| 38 | try: |
| 39 | # Si aucune base n'est spécifiée, on se connecte sans base |
| 40 | db_part = f"/{database}" if database else "" |
| 41 | connection_str = f"mysql+pymysql://{username}:{password}@{ip}:3306{db_part}" |
| 42 | engine = create_engine(connection_str, connect_args={"connect_timeout": 10}) |
| 43 | self.sql_connected = True |
| 44 | logger.info(f"Connected to {ip} via SQL with username {username}" + (f" to database {database}" if database else "")) |
| 45 | return engine |
| 46 | except Exception as e: |
| 47 | logger.error(f"SQL connection error for {ip} with user '{username}' and password '{password}'" + (f" to database {database}" if database else "") + f": {e}") |
| 48 | return None |
| 49 | |
| 50 | def find_tables(self, engine): |
| 51 | """ |