主要需求从一个 MySQL 数据库（源数据库）同步表结构到另一个 MySQL 数据库（目标数据库）。具体而言，它会从源数据库中获取每个表的创建 SQL 语句，并在目标数据库中执行这些语句，以创建相同的表结构。不依赖navicat可以用下面脚本很方便的进行同步
步骤分解
-
导入必要的库：
import mysql.connector这里导入了
mysql.connector，这是一个用于连接 MySQL 数据库的 Python 库。 -
数据库连接配置：
source_db_config = {'host': 'xxx','port': 32562,'user': 'root','password': 'xxx','database': 'xxx' }target_db_config = {'host': 'xxx','port': 30653,'user': 'root','password': 'xxx','database': 'xxx' }- 这里定义了两个字典
source_db_config和target_db_config，分别包含源数据库和目标数据库的连接信息。 - 包括主机地址、端口、用户名、密码和数据库名称。
- 这里定义了两个字典
-
获取表结构的函数：
def get_table_structure(cursor, table_name):cursor.execute(f"SHOW CREATE TABLE {table_name};")result = cursor.fetchone()return result[1] # 获取创建表的 SQL 语句- 该函数接受一个数据库游标和表名作为参数。
- 使用
SHOW CREATE TABLESQL 语句获取指定表的创建 SQL 语句，并返回该语句。
-
同步表结构的函数：
def sync_table_structure(source_cursor, target_cursor, table_name):create_table_sql = get_table_structure(source_cursor, table_name)try:target_cursor.execute(f"DROP TABLE IF EXISTS {table_name};") # 可选：先删除目标表target_cursor.execute(create_table_sql)print(f"表 {table_name} 结构已同步。")except mysql.connector.Error as err:print(f"表 {table_name} 同步失败: {err}")- 此函数用于从源数据库获取表结构并在目标数据库中创建相同的表。
- 首先调用
get_table_structure获取创建表的 SQL 语句。 - 然后执行
DROP TABLE IF EXISTS以删除目标数据库中已存在的同名表（这一操作是可选的）。 - 最后，使用获取的 SQL 语句在目标数据库中创建表，并在控制台输出同步状态。
- 如果在执行过程中出现错误，会捕获并打印错误信息。
-
主逻辑函数：
def main():# 连接源数据库source_conn = mysql.connector.connect(**source_db_config)source_cursor = source_conn.cursor()# 连接目标数据库target_conn = mysql.connector.connect(**target_db_config)target_cursor = target_conn.cursor()# 获取源数据库中的所有表source_cursor.execute("SHOW TABLES;")tables = source_cursor.fetchall()for (table_name,) in tables:sync_table_structure(source_cursor, target_cursor, table_name)# 关闭连接source_cursor.close()source_conn.close()target_cursor.close()target_conn.close()main函数是程序的入口点。- 首先，连接源和目标数据库，并创建相应的游标。
- 使用
SHOW TABLES;查询获取源数据库中的所有表名。 - 对于每个表，调用
sync_table_structure进行结构同步。 - 最后，关闭所有游标和连接以释放资源。
完整脚本
import mysql.connector# MySQL 配置
source_db_config = {'host': '192.168.110.xx','port': 32562,'user': 'root','password': 'xxx','database': 'xxx'
}target_db_config = {'host': '192.168.110.xxx2','port': 30653,'user': 'root','password': 'xx','database': 'xx'
}def get_table_structure(cursor, table_name):cursor.execute(f"SHOW CREATE TABLE {table_name};")result = cursor.fetchone()return result[1] # 获取创建表的 SQL 语句def sync_table_structure(source_cursor, target_cursor, table_name):create_table_sql = get_table_structure(source_cursor, table_name)try:# 在目标数据库中执行创建表的 SQLtarget_cursor.execute(f"DROP TABLE IF EXISTS {table_name};") # 可选：先删除目标表target_cursor.execute(create_table_sql)print(f"表 {table_name} 结构已同步。")except mysql.connector.Error as err:print(f"表 {table_name} 同步失败: {err}")def main():# 连接源数据库source_conn = mysql.connector.connect(**source_db_config)source_cursor = source_conn.cursor()# 连接目标数据库target_conn = mysql.connector.connect(**target_db_config)target_cursor = target_conn.cursor()# 获取源数据库中的所有表source_cursor.execute("SHOW TABLES;")tables = source_cursor.fetchall()for (table_name,) in tables:sync_table_structure(source_cursor, target_cursor, table_name)# 关闭连接source_cursor.close()source_conn.close()target_cursor.close()target_conn.close()if __name__ == "__main__":main()后面就不赘述同步完检查了
