Edit: /opt/dedrads/innodb_converter.py (9343B)
#!/usr/lib/rads/venv/bin/python3
"""Converts all MyISAM tables to InnoDB"""
# import list type
from pathlib import Path
import os
import sys
from argparse import ArgumentParser
import logging
import pymysql
# check for root
if os.getuid() != 0:
sys.exit('This script must be run as root')
logger = logging.getLogger('innodb_convert.py')
# log to /var/log/innodb_convert.log
logger.setLevel(logging.DEBUG)
# print info and above to log file and console
console_handler = logging.StreamHandler(sys.stderr)
console_handler.setLevel(logging.INFO)
logger.addHandler(console_handler)
file_handler = logging.FileHandler('/var/log/innodb_convert.log')
file_handler.setLevel(logging.DEBUG)
logger.addHandler(file_handler)
def confirm_backups(check: bool, default='N'):
"""Confirms that backups have been taken, with configurable default (Y or N)"""
if check:
return
prompt = f"Have you taken backups? (y/{default}) "
answer = input(prompt).strip() or default
if answer.lower() != 'y':
prompt2 = f"Do you wish to continue anyway? (y/{default}) "
answer2 = input(prompt2).strip() or default
if answer2.lower() != 'y':
sys.exit('Exiting...')
def get_db_config() -> dict:
"""Returns a dictionary of the MySQL configuration"""
mysql_cnf = Path('/root/.my.cnf')
if not mysql_cnf.exists():
logger.error('Error: /root/.my.cnf does not exist')
sys.exit(1)
my_cnf_data = {}
with open(mysql_cnf, encoding='utf-8') as file:
for line in file:
if line.startswith('['):
section = line.strip('[]\n')
my_cnf_data[section] = {}
else:
key, value = line.strip().split('=')
value = value.replace("'", "").replace('"', '')
my_cnf_data[section][key] = value
db_config_dict = {
'host': 'localhost',
'user': my_cnf_data['client']['user'],
'password': my_cnf_data['client']['password'],
'charset': 'utf8mb4',
'cursorclass': pymysql.cursors.DictCursor,
}
return db_config_dict
db_config = get_db_config()
def list_databases() -> list:
"""Returns a list of all databases"""
databases = []
try:
with pymysql.connect(**db_config) as connection:
with connection.cursor() as cursor:
cursor.execute('SHOW DATABASES')
databases = cursor.fetchall()
databases = [x['Database'] for x in databases]
return databases
except pymysql.err.MySQLError as e:
error = f'Error: {e}'
sys.exit(error)
def list_tables(database: str) -> tuple[str | None, list[str] | None]:
"""Returns a list of all tables in a database"""
tables = []
try:
with pymysql.connect(database=database, **db_config) as connection:
with connection.cursor() as cursor:
cursor.execute('SHOW TABLES')
tables = cursor.fetchall()
tables = [x['Tables_in_' + database] for x in tables]
return None, tables
except pymysql.err.MySQLError as e:
error = f'Error on {database}: {e}'
logger.error(error)
return error, None
def check_engine(database: str, table: str) -> tuple[str | None, str | None]:
"""Returns the engine of a table"""
try:
with pymysql.connect(database=database, **db_config) as connection:
with connection.cursor() as cursor:
error = None
cursor.execute('SHOW TABLE STATUS LIKE %s', (table,))
table_status = cursor.fetchone()
return error, table_status['Engine']
except pymysql.err.MySQLError as e:
error = f'Error on {database}.{table}: {e}'
logger.error(error)
return error, None
def convert_table(database: str, table: str, target) -> str | None:
"""Converts a table to InnoDB to the target engine(InnoDB or MyISAM)"""
try:
with pymysql.connect(database=database, **db_config) as connection:
with connection.cursor() as cursor:
cursor.execute(f'ALTER TABLE {table.replace("`", "``")} ENGINE={target}')
return None
except pymysql.err.MySQLError as e:
error = f'Error: {e}'
logger.error(error)
return error
def get_mysql_innodb_settings() -> tuple[list, dict]:
"""Returns the innodb settings from my.cnf and my.cnf.d"""
conf_dir = Path('/etc/my.cnf.d')
conf_files = list(conf_dir.glob('*.cnf'))
main_conf = Path('/etc/my.cnf')
innodb_settings = {}
key_errors = []
conf_files.append(main_conf)
for file in conf_files:
innodb_settings[str(file.name)] = {}
if file.exists():
with file.open('r', encoding='utf-8') as f:
for line in f:
if line.startswith('innodb_'):
try:
key, value = line.strip().split('=')
key = key.strip()
except ValueError:
key = line.strip()
value = ''
for _, settings in innodb_settings.items():
if key in settings:
error = (
f'Error: {key} is defined in multiple files'
)
key_errors.append(error)
innodb_settings[str(file.name)][key] = value
return key_errors, innodb_settings
def clean_db_list(databases: list) -> list:
"""Removes system databases from the list"""
skips = ['information_schema', 'performance_schema', 'mysql', 'sys', 'test']
for database in skips:
if database in databases:
databases.remove(database)
return databases
def process_databases(databases: list, args) -> list:
"""Processes all databases"""
errors = []
if args.myisam:
current='InnoDB'
target='MyISAM'
else:
current='MyISAM'
target='InnoDB'
for database in databases:
error, tables = list_tables(database)
if args.table:
if args.table in tables:
tables = [args.table]
else:
error = f'Error: {args.table} does not exist in {database}'
logger.error(error)
errors.append(error)
continue
if error:
errors.append(error)
continue
for table in tables:
error, engine = check_engine(database, table)
if error:
errors.append(error)
continue
if args.check:
logger.info('%s.%s is %s', database, table, engine)
continue
if engine.lower() == current.lower():
logger.info('Converting %s.%s to %s', database, table, target)
error = convert_table(database, table, target)
if error:
errors.append(error)
continue
return errors
def main():
"""Main function"""
parser = ArgumentParser(description='Converts MyISAM tables to InnoDB')
parser.add_argument('-a', '--all', help='Select all databases', action='store_true')
parser.add_argument('-d', '--database', help='Database to convert')
parser.add_argument('-t', '--table', help='Specify a table to convert. combine with -d')
parser.add_argument('--check', help='Check selected databases', action='store_true')
parser.add_argument('--myisam', help='Convert to MyISAM instead of InnoDB', action='store_true')
args = parser.parse_args()
if args.all and args.database:
logger.error('Error: --all and --database are mutually exclusive')
parser.print_help()
sys.exit(1)
elif not args.all and not args.database:
logger.error('Error: No database specified')
logger.error('must use either --all (-a) or --database (-d) flags')
parser.print_help()
sys.exit(1)
elif args.table and not args.database:
parser.error('--table requires --database')
errors = []
confirm_backups(args.check)
databases = list_databases()
databases = clean_db_list(databases)
if args.database:
database = args.database
if database not in databases:
error = f'Error: {database} does not exist or is a system database'
logger.error(error)
sys.exit(error)
databases = [database]
errors = process_databases(databases, args)
key_errors, innodb_settings = get_mysql_innodb_settings()
if len(errors) > 0 or len(key_errors) > 0:
logger.error('Errors occurred:')
for error in errors:
logger.error("\t%s", error)
for key_error in key_errors:
logger.error("\t%s", key_error)
logger.info('InnoDB settings:')
for filename, settings in innodb_settings.items():
logger.info('%s', filename)
for key, value in settings.items():
if value == '':
logger.info('\t%s', key)
else:
logger.info('\t%s = %s', key, value)
if __name__ == '__main__':
main()