ISEE小语
在网上看到马东在《奇葩说》中曾说的一句话：“一个人的情绪到了尽头，是沉默。”
在某个时候，沉默只是想放过自己！
回顾上篇
python中不常见却强大并好用的6个三方库，精简版
ISEE小栈，公众号：ISEE小栈python中不常见却强大并好用的6个三方库，精简版
开始本篇
导出数据库的表结构和数据，咱们经常使用的方式是在工具中直接导出，如在Navicat中[转储SQL文件]——>[结构和数据]，可以很直接的导出sql文件。
那么有同学就会问到，直接用Navicat等这样的工具不就可以了么，为什么还要费劲的去用代码实现？
其实本次实际运用的场景是将正式环境的或是总库的数据库，快速地并批量的导出数据表结构和少量的数据，创建到新的数据库中，为测试自动化提供基础数据。
所以，这个是期初是和测试自动化结合使用的。也适用于运维备份数据库。总之开发、测试、运维、数据分析等有需用之。
此次实现功能具有特殊性和定制性，不同的数据库有可能需要有些特殊地处理。有借用到自己数据库的同学，如果有不成功的情况，那就需要根据实际情况在这基础上稍作优化。
环境：
Pycharm
Python 3.9.18
安装：
pip install mysql-connector-python==8.1.0
pip install PyMySQL==1.1.0
导入：
import mysql.connector
import datetime
导出数据库的表结构和数据，分为三部分：
第一部分 生成删除表的SQL
第二部分 生成表的结构
第三部分 导出数据
在实现以上三个部分前，要先连接MySql原数据库，并获取数据库中所有的表名。
原数据库表：
连接MySql：
# 连接MySqldb_config = {'host': '127.0.0.1','user': 'root','password': '123456','port': 3306,'database': 'self_test'}cnx = mysql.connector.connect(**db_config)cursor = cnx.cursor()# 查询所有表名cursor.execute("SHOW TABLES")table_names = cursor.fetchall()print(table_names)
结果：
看到把数据库表都输出了，说明已经连接成功~！
那么接下来就开始……
第一部分
生成删除表的SQL
首要任务是生成删除表的SQL，防止在新库创建表的时候，有相同的表会有冲突的情况。
类似于Navicat工具，导出的SQL，删除表的操作也是在前面
for table_name in table_names:table_name = table_name[0]drop_sql = f"DROP TABLE IF EXISTS `{table_name}`;\n"print(drop_sql)
（左右滑动查看完整代码）
结果：
第二部分
生成表的结构
生成表的结构，我们需要：
查询表结构信息，包括字段名、数据类型和注释
查询主键信息
查询索引信息
查询表的字符集和排序规则
首先，组合表结构的字段名、数据类型和注释
for table_name in table_names:table_name = table_name[0]# 查询表结构信息，包括字段名、数据类型和注释cursor.execute(f"SHOW FULL COLUMNS FROM {table_name}")columns = cursor.fetchall()# 查询索引信息cursor.execute(f"SHOW INDEX FROM {table_name}")indexes = cursor.fetchall()# 查询主键信息cursor.execute(f"SHOW KEYS FROM {table_name} WHERE Key_name = 'PRIMARY'")primary_keys = cursor.fetchall()# 查询表的字符集和排序规则cursor.execute(f"SHOW TABLE STATUS LIKE '{table_name}'")table_info = cursor.fetchone()# 首先，组合表结构的字段名、数据类型和注释create_sql = ""create_ = f"CREATE TABLE `{table_name}` (\n"create_sql += create_for column in columns:column_name = column[0] # 字段名data_type = column[1] # 字段类型charset = column[2] # 字符集Collation = f"CHARACTER SET {charset.split('_')[0]} COLLATE {charset}" if charset is not None else ''if column[3] == 'NO':is_Null = 'NOT NULL'if column[5] == '':Default = "DEFAULT ''"elif column[5] is not None:Default = f"DEFAULT '{column[5]}'"else:Default = ''else:is_Null = 'NULL'if column[5] is None:Default = 'DEFAULT NULL'elif column[5] == '':Default = "DEFAULT ''"elif column[5] == 'CURRENT_TIMESTAMP': # 判断时间类型的默认值Default = "DEFAULT CURRENT_TIMESTAMP(0) ON UPDATE CURRENT_TIMESTAMP(0)"else:Default = f"DEFAULT '{column[5]}'"# 字段注释comment = column[8]# 判断sql字段最后一下后面是否有主键或和索引，如果没有，最后一个字段不加","号if column == columns[-1] and len(indexes) == 0 and len(primary_keys) == 0:column_ = f" `{column_name}` {data_type} {Collation} {is_Null} {Default} COMMENT '{comment}'\n"create_sql += column_else:column_ = f" `{column_name}` {data_type} {Collation} {is_Null} {Default} COMMENT '{comment}',\n"create_sql += column_print(create_sql)
（左右滑动查看完整代码）
结果：
从以上结果可以看出，表中有主键或索引等信息，那么接下来把这些也补充上
主键，如果只有主键，添加完成后则结束；如果还有索引，则继续将索引等信息也添加
# 如果只有主键，则结束if len(indexes) == 1:if len(primary_keys) > 0:primary_key_columns = [pk[4] for pk in primary_keys]primary_btree_columns = [pk[10] for pk in primary_keys]primary_ = f" PRIMARY KEY ({', '.join(primary_key_columns)}) USING {', '.join(set(primary_btree_columns))}\n"create_sql += primary_# 如果还有索引，则继续添加else:if len(primary_keys) > 0:primary_key_columns = [pk[4] for pk in primary_keys]primary_btree_columns = [pk[10] for pk in primary_keys]primary_ = f" PRIMARY KEY ({', '.join(f'`{pkc}`' for pkc in primary_key_columns)}) USING {', '.join(set(primary_btree_columns))},\n"create_sql += primary_# 添加索引index_dict = {}for index in indexes:index_name = index[2]column_name = index[4]column_btree = index[10]if index_name not in index_dict:index_dict[index_name] = []index_dict[index_name].append(column_name)# 获取所有的键keys = list(index_dict.keys())for index_name in list(index_dict.keys()):columns = index_dict[index_name]if index_name == 'PRIMARY':continue# 如果索引不是最后一个，在输出后面添加逗号if index_name != keys[-1]:index_ = f" INDEX `{index_name}` ({', '.join(f'`{iname}`' for iname in columns)}) USING {column_btree},\n"create_sql += index_# 如果索引是最后一个，不添加逗号else:index_ = f" INDEX `{index_name}` ({', '.join(f'`{iname}`' for iname in columns)}) USING {column_btree}\n"create_sql += index_print(create_sql)
（左右滑动查看完整代码）
结果：
以上结果，我们将主键信息和索引信息都获取到了。
那么最后需要获取数据表的引擎、字符集、表注释等信息了
# 获取引擎、字符集、表注释等ENGINE = table_info[1]AUTO_INCREMENT = '' if table_info[10] is None else f"AUTO_INCREMENT = {table_info[10]}"charset_table = table_info[14] # 表字符集charset_table_lst = charset_table.split('_')CHARACTER = f'CHARACTER SET = {charset_table_lst[0]} COLLATE = {charset_table}'row_format = table_info[16]COMMENT = '' if table_info[17] == '' else f"COMMENT = '{table_info[17]}'" # 表注释engine_ = f") ENGINE = {ENGINE} {AUTO_INCREMENT} {CHARACTER} {COMMENT} {row_format};\n"create_sql += engine_print(create_sql)
（左右滑动查看完整代码）
结果：
以上创建表的SQL已成，接下来测试一下，我们新建个数据库，名字随意，在Navicat上将以上SQL执行一下
OK，成功
第三部分
导出数据
上面导出表结构是备份数据库的必要操作，这部分导出数据是可选操作，如果只想导出数据库结构，这部分就不需要添加了。
本次原计划是导出每个表中最新的三条数据，来做基础数据。
换个直白的说法，就是查询最新的3条数据，生成insert语句。
这个查询的sql，可以是全部数据，也可以根据条件查询，总之用的时候自定义查询的sql语句即可。
for table_name in table_names:table_name = table_name[0]# 查询最新的3条数据query = f"SELECT * FROM {table_name} order by id desc limit 3"cursor.execute(query)for row in cursor:values_list = []for value in row:if isinstance(value, str):values_list.append(f"'{str(value)}'")elif value is None:values_list.append('NULL')elif isinstance(value, datetime.datetime):values_list.append(f"'{str(value.strftime('%Y-%m-%d %H:%M:%S'))}'")else:values_list.append(str(value))# 使用逗号连接列表中的所有元素，得到最终的分隔字符串values_str = ','.join(values_list)insert_sql = f"INSERT INTO {table_name} VALUES ({values_str});\n"print(insert_sql)
（左右滑动查看完整代码）
结果：
已生成insert语句，接着第二部分创建的库和表，我们在Navicat中进行测试，看是否可以成功插入
OK，成功
总结
统一归整与输出到文件
为了更方便的阅读，以上是分步记录的，最后会调整一下代码位置，按数据表结构更完整的输出来，并保存到新建的.sql文件中。方便后续自动化执行使用。
统一归整后的代码结构：
输出到SQL文件：
大家有没有发现，最终导出的这个sql文件内容，和Navicat中导出的格式相差无几？
笔者把实例中使用的数据库和源码都共享了，有兴趣者，可试之！
后台回复“export_mysql”即可获取~！
寄语：世间三美，明月，清风，眼前……
看到这儿的朋友帮点个“赞”和“在看”，谢谢支持~！
文章就分享到这儿，喜欢就点个赞吧!
