Edit: /opt/dedrads/mysql/mysqlrepair.py (1939B)
#!/usr/lib/rads/venv/bin/python3
"""repairs database tables"""
from argparse import ArgumentParser
from pathlib import Path
import sys
from typing import Literal
import pymysql
from pymysql.cursors import Cursor
# specify the socket file for pymysql
if Path("/etc/redhat-release").exists():
SOCK = "/var/lib/mysql/mysql.sock"
else:
# ubuntu support
SOCK = "/run/mysqld/mysqld.sock"
def repair_tables(
conn: pymysql.Connection, dbcmd: Literal['OPTIMIZE', 'REPAIR']
) -> int:
rcode = 0
with conn.cursor(Cursor) as cur:
try:
cur.execute("SHOW TABLES")
tablenames = [x[0] for x in cur.fetchall()]
except pymysql.Error as exc:
sys.exit(exc)
for table in tablenames:
escaped = str(table).replace('`', '``')
try:
cur.execute(f"{dbcmd} TABLE `{escaped}`")
print(*cur.fetchone())
except pymysql.Error as exc:
print(
f"Error running {dbcmd} TABLE on {table}: {exc}",
file=sys.stderr,
)
rcode = 2
continue
return rcode
def parse_args() -> tuple[Literal['OPTIMIZE', 'REPAIR'], str]:
parser = ArgumentParser(description=__doc__)
dbcmd = parser.add_mutually_exclusive_group(required=True)
dbcmd.add_argument(
'--repair', action='store_const', const='REPAIR', dest='dbcmd'
)
dbcmd.add_argument(
'--optimize', action='store_const', const='OPTIMIZE', dest='dbcmd'
)
parser.add_argument('dbname')
args = parser.parse_args()
return args.dbcmd, args.dbname
def main():
dbcmd, dbname = parse_args()
with pymysql.connect(
read_default_file="/root/.my.cnf",
unix_socket=SOCK,
database=dbname,
) as conn:
rcode = repair_tables(conn, dbcmd)
sys.exit(rcode)
if __name__ == '__main__':
main()