一、生成建表语句

• 使用Python批量自动生成Mysql中已有表的建表语句

import osdef generate_create_table_sql(cursor,file_path,table_name,schema):# 获取表的列信息cursor.execute(f"SHOW FULL COLUMNS FROM {table_name}")columns = cursor.fetchall()# 获取表的主键信息cursor.execute(f"SHOW KEYS FROM {table_name} WHERE Key_name = 'PRIMARY'")primary_keys = cursor.fetchall()#获取表的唯一约束信息cursor.execute(f"SHOW INDEX FROM {table_name} WHERE Non_unique = 0 AND Key_name != 'PRIMARY'")unique_constraints = cursor.fetchall()#获取表的索引信息cursor.execute(f"SHOW INDEX FROM {table_name} WHERE Key_name != 'PRIMARY' AND Non_unique != 0")indexes = cursor.fetchall()#获取表注释信息cursor.execute(f"SHOW TABLE STATUS LIKE '{table_name}'")table_status = cursor.fetchone()table_comment = table_status[17]# 生成建表语句create_table_sql = f"CREATE TABLE IF NOT EXISTS {schema}.{table_name} (\n"for column in columns:column_name = column[0]data_type = column[1]is_nullable = column[3]column_default = column[5]column_comment = column[8]# 判断是否为主键is_primary_key = column_name in [pk[4] for pk in primary_keys]# 构建列的字符串表示column_definition = f"    {column_name} {data_type}"if column_name.upper() == 'ID':column_definition += " AUTO_INCREMENT"# 添加是否为空约束if is_nullable == 'NO':column_definition += " NOT NULL"# 添加默认值if column_default is not None:column_definition += f" DEFAULT {column_default}"# 添加注释if column_comment:column_definition += f" COMMENT '{column_comment}'"# 添加主键约束if is_primary_key:column_definition += " PRIMARY KEY"create_table_sql += f"{column_definition},\n"#添加唯一约束,若唯一约束名相同,唯一约束应拼接在一起unique_constraints_dict = {}for constraint in unique_constraints:constraint_name = constraint[2]column_name = constraint[4]if constraint_name in unique_constraints_dict:unique_constraints_dict[constraint_name].append(column_name)else:unique_constraints_dict[constraint_name] = [column_name]for constraint_name,column_names in unique_constraints_dict.items():columns_str = ",".join(column_names)create_table_sql += f"    CONSTRAINT {constraint_name} unique ({columns_str}),\n"#添加索引indexes_dict = {}for index in indexes:index_name = index[2]column_name = index[4]if index_name in indexes_dict:indexes_dict[index_name].append(column_name)else:indexes_dict[index_name] = [column_name]for index_name,column_names in indexes_dict.items():columns_str = ",".join(column_names)create_table_sql += f"    INDEX {index_name} ({columns_str}),\n"# 去除最后一行的逗号create_table_sql = create_table_sql.rstrip(",\n")create_table_sql += f"\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT = '{table_comment}';"file_path = os.path.join(file_path,f"{table_name}.sql")with open(file_path,'w',encoding="utf") as f:f.write(create_table_sql)print(f"Successfully write {table_name}")

import pymysql
from Mysql.GenerateCreateTableSql import generate_create_table_sqlconn = pymysql.connect(host='localhost',port=1521,user='test',password='test',database='test_db',charset='utf8mb4')
cursor = conn.cursor()
table_names = []
schema = 'test_db'
file_path = r'D:/data'
with open(r'table_names.txt','r') as f:table_names = [line.strip() for line in f if line.strip()]
for tab in table_names:generate_create_table_sql(cursor,file_path,tab,schema)# 关闭数据库连接
cursor.close()
conn.close()

二、执行建表语句

import os
import pymysql#连接数据库
conn = pymysql.connect(host='localhost',port=1521,user='test',password='test',database='test_db',charset='utf8mb4')cursor = conn.cursor()#读取并执行sql脚本文件
script_file = r'D:\data'
for file in os.listdir(script_file):if file.endswith('.sql'):script_path = os.path.join(script_file,file)with open(script_path,'r',encoding='utf-8') as f:script = f.read()cursor.execute(script)conn.commit()print(f'successfully executed {file}')#关闭连接
cursor.close()
conn.close()print("Sql脚本执行完成")