python 比较 mysql 表结构差异

最近在做项目的时候，需要比对两个数据库的表结构差异，由于表数量比较多，人工比对的话需要大量时间，且不可复用，于是想到用 python 写一个脚本来达到诉求，下次有相同诉求的时候只需改 sql 文件名即可。

compare_diff.py:

import re
import json# 建表语句对象
class TableStmt(object):table_name = ""create_stmt = ""# 表对象
class Table(object):table_name = ""fields = []indexes = []# 字段对象
class Field(object):field_name = ""field_type = ""# 索引对象
class Index(object):name = ""type = ""columns = ""# 自定义JSON序列化器，非必须，打印时可用到
def obj_2_dict(obj):if isinstance(obj, Field):return {"field_name": obj.field_name,"field_type": obj.field_type}elif isinstance(obj, Index):return {"name": obj.name,"type": obj.type,"columns": obj.columns}raise TypeError(f"Type {type(obj)} is not serializable")# 正则表达式模式来匹配完整的建表语句
create_table_pattern = re.compile(r"CREATE TABLE `(?P<table_name>\w+)`.*?\)\s*ENGINE[A-Za-z0-9=_ ''\n\r\u4e00-\u9fa5]+;",re.DOTALL | re.IGNORECASE
)# 正则表达式模式来匹配字段名和字段类型，只提取基本类型忽略其他信息
table_pattern = re.compile(r"^\s*`(?P<field>\w+)`\s+(?P<type>[a-zA-Z]+(?:\(\d+(?:,\d+)?\))?)",re.MULTILINE
)# 正则表达式模式来匹配索引定义
index_pattern = re.compile(r'(?<!`)KEY\s+`?(\w+)`?\s*\(([^)]+)\)|'r'PRIMARY\s+KEY\s*\(([^)]+)\)|'r'UNIQUE\s+KEY\s+`?(\w+)`?\s*\(([^)]+)\)|'r'FULLTEXT\s+KEY\s+`?(\w+)`?\s*\(([^)]+)\)',re.IGNORECASE)# 提取每个表名及建表语句
def extract_create_table_statements(sql_script):matches = create_table_pattern.finditer(sql_script)table_create_stmts = []for match in matches:tableStmt = TableStmt()tableStmt.table_name = match.group('table_name').lower()  # 表名统一转换成小写tableStmt.create_stmt = match.group(0).strip()  # 获取匹配到的整个建表语句table_create_stmts.append(tableStmt)return table_create_stmts# 提取索引
def extract_indexes(sql):matches = index_pattern.findall(sql)indexes = []for match in matches:index = Index()if match[0]:  # 普通索引index.type = 'index'index.name = match[0].lower()index.columns = match[1].lower()elif match[2]:  # 主键index.type = 'primary key'index.name = 'primary'index.columns = match[2].lower()elif match[3]:  # 唯一索引index.type = 'unique index'index.name = match[3].lower()index.columns = match[4].lower()elif match[5]:  # 全文索引index.type = 'fulltext index'index.name = match[5].lower()index.columns = match[6].lower()indexes.append(index)return indexes# 提取字段
def extract_fields(sql):matches = table_pattern.finditer(sql)fields = []for match in matches:field = Field()field.field_name = match.group('field').lower()  # 字段名统一转换成小写field.field_type = match.group('type').lower()  # 字段类型统一转换小写fields.append(field)return fields# 提取表信息
def extract_table_info(tableStmt: TableStmt):table = Table()table.table_name = tableStmt.table_name.lower()# 获取字段table.fields = extract_fields(tableStmt.create_stmt)# 获取索引table.indexes = extract_indexes(tableStmt.create_stmt)return table# 提取sql脚本中所有的表
def get_all_tables(sql_script):table_map = {}table_stmts = extract_create_table_statements(sql_script)for stmt in table_stmts:table = extract_table_info(stmt)table_map[table.table_name] = tablereturn table_map# 比较两个表的字段
def compare_fields(source: Table, target: Table):source_fields_map = {field.field_name: field for field in source.fields}target_fields_map = {field.field_name: field for field in target.fields}source_fields_not_in_target = []fields_type_not_match = []#  source表有，而target表没有的字段for field in source.fields:if field.field_name not in target_fields_map.keys():source_fields_not_in_target.append(field.field_name)continuetarget_field = target_fields_map.get(field.field_name)if field.field_type != target_field.field_type:fields_type_not_match.append("field=" + field.field_name + ", source type: " + field.field_type + ", target type: " + target_field.field_type)target_fields_not_in_source = []#  target表有，而source表没有的字段for field in target.fields:if field.field_name not in source_fields_map.keys():target_fields_not_in_source.append(field.field_name)continue# 不用再比较type了，因为如果这个字段在source和target都有的话，前面已经比较过type了return source_fields_not_in_target, fields_type_not_match, target_fields_not_in_source# 比较两个表的索引
def compare_indexes(source: Table, target: Table):source_indexes_map = {index.name: index for index in source.indexes}target_indexes_map = {index.name: index for index in target.indexes}source_indexes_not_in_target = []index_column_not_match = []index_type_not_match = []for index in source.indexes:if index.name not in target_indexes_map.keys():# source表有而target表没有的索引source_indexes_not_in_target.append(index.name)continuetarget_index = target_indexes_map.get(index.name)# 索引名相同，类型不同if index.type != target_index.type:index_type_not_match.append("name=" + index.name + ", source type: " + index.type + ", target type: " + target_index.type)continue# 索引名和类型都相同，字段不同if index.columns != target_index.columns:index_column_not_match.append("name=" + index.name + ", source columns=" + index.columns + ", target columns=" + target_index.columns)target_indexes_not_in_source = []for index in target.indexes:if index.name not in source_indexes_map.keys():# target表有而source表没有的索引target_indexes_not_in_source.append(index.name)continuereturn source_indexes_not_in_target, index_column_not_match, index_type_not_match, target_indexes_not_in_source# 打印比较的结果，如果结果为空列表（说明没有不同）则不打印
def print_diff(desc, compare_result):if len(compare_result) > 0:print(f"{desc} {compare_result}")# 比较脚本里面的所有表
def compare_table(source_sql_script, target_sql_script):source_table_map = get_all_tables(source_sql_script)target_table_map = get_all_tables(target_sql_script)source_table_not_in_target = []for key, source_table in source_table_map.items():# 只比较白名单里面的表if len(white_list_tables) > 0 and key not in white_list_tables:continue# 不比较黑名单里面的表if len(black_list_tables) > 0 and key in black_list_tables:continueif key not in target_table_map.keys():# source有而target没有的表source_table_not_in_target.append(key)continuetarget_table = target_table_map[key]# 比较字段(source_fields_not_in_target, fields_type_not_match, target_fields_not_in_source) = compare_fields(source_table, target_table)# 比较索引(source_indexes_not_in_target, index_column_not_match, index_type_not_match, target_indexes_not_in_source) = compare_indexes(source_table, target_table)print(f"====== table = {key} ======")print_diff("source field not in target, fields:", source_fields_not_in_target)print_diff("target field not in source, fields:", target_fields_not_in_source)print_diff("field type not match:", fields_type_not_match)print_diff("source index not in target, indexes:", source_indexes_not_in_target)print_diff("target index not in source, indexes:", target_indexes_not_in_source)print_diff("index type not match:", index_type_not_match)print_diff("index column not match:", index_column_not_match)print("")# 找出target有而source没有的表target_table_not_in_source = []for key, target_table in target_table_map.items():# 只比较白名单里面的表if len(white_list_tables) > 0 and key not in white_list_tables:continue# 不比较黑名单里面的表if len(black_list_tables) > 0 and key in black_list_tables:continueif key not in source_table_map.keys():target_table_not_in_source.append(key)print_diff("source table not in target, table list:", source_table_not_in_target)print_diff("target table not in source, table list:", target_table_not_in_source)# 读取sql文件
def sql_read(file_name):with open(file_name, "r", encoding='utf-8') as file:return file.read()def print_all_tables():table_map = get_all_tables(sql_read("sql1.sql"))for key, item in table_map.items():print(key)print(json.dumps(item.fields, default=obj_2_dict, ensure_ascii=False, indent=4))print(json.dumps(item.indexes, default=obj_2_dict, ensure_ascii=False, indent=4))print("")# print_all_tables()# 黑白名单设置，适用于只比较所有表中一部分表的情况
# 白名单表，不为空的话，只比较这里面的表
white_list_tables = []
# 黑名单表，不为空的话，不比较这里面的表
black_list_tables = []if __name__ == '__main__':# 说明：mysql默认大小写不敏感，如果数据库设置了大小写敏感，脚本需要修改，里面所有的表名、字段名和索引名都默认转了小写再去比较的source_script = sql_read("sql1.sql")target_script = sql_read("sql2.sql")compare_table(source_script, target_script)

运行效果如下：

====== table = table1 ======
source field not in target, fields: ['age', 'email']
target field not in source, fields: ['name']
field type not match: ['field=created_at, source type: date, target type: bigint(20)', 'field=updated_at, source type: timestamp, target type: date']
source index not in target, indexes: ['unique_name']
target index not in source, indexes: ['idx_country_env']====== table = table2 ======
index type not match: ['name=fulltext_index, source type: fulltext index, target type: index']
index column not match: ['name=index, source columns=`age`, target columns=`description`']====== table = table3 ======
index column not match: ['name=primary, source columns=`id`, `value`, target columns=`value`, `id`']source table not in target, table list: ['activity_instance']
target table not in source, table list: ['table5']

结果说明：

  • 按照 table 来打印 source table 和 target table 的字段和索引差异，此时 table 在两个 sql 脚本里都存在
  • 最后打印只在其中一个 sql 脚本里存在的 table list

sql1.sql:

CREATE TABLE `table1` (`id` INT(11) NOT NULL AUTO_INCREMENT,`age` INT(11) DEFAULT NULL,`email` varchar(32)   DEFAULT NULL COMMENT '邮箱',`created_at` date DEFAULT NULL,`updated_at` TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`),UNIQUE KEY `unique_name` (`name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT ='测试表';CREATE TABLE `table2` (`id` INT(11) NOT NULL,`description` TEXT NOT NULL,`created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,PRIMARY KEY (`id`),UNIQUE KEY `unique_name` (`name`),KEY `index` (`age`),FULLTEXT KEY `fulltext_index` (`name`, `age`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;CREATE TABLE `table3` (`id` INT(11) NOT NULL AUTO_INCREMENT,`value` DECIMAL(10,2) NOT NULL,`updated_at` TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`, `value`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;/******************************************/
/*   DatabaseName = database   */
/*   TableName = activity_instance   */
/******************************************/
CREATE TABLE `activity_instance`
(`id`                   bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',`gmt_create`           bigint(20) NOT NULL COMMENT '创建时间',`gmt_modified`         bigint(20) NOT NULL COMMENT '修改时间',`activity_name`        varchar(400)  NOT NULL COMMENT '活动名称',`benefit_type`         varchar(16)   DEFAULT NULL,`benefit_id`           varchar(32)   DEFAULT NULL,PRIMARY KEY (`id`),KEY `idx_country_env` (`env`, `country_code`),KEY `idx_benefit_type_id` (`benefit_type`, `benefit_id`)
) ENGINE = InnoDBAUTO_INCREMENT = 139DEFAULT CHARSET = utf8mb4 COMMENT ='活动时间模板表'
;

sql2.sql:

CREATE TABLE `TABLE1` (`id` INT(11) NOT NULL AUTO_INCREMENT,`name` VARCHAR(255) NOT NULL,`created_at` bigint(20) DEFAULT NULL,`updated_at` date ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`),KEY `idx_country_env` (`env`, `country_code`),
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT ='测试表';CREATE TABLE `table2` (`id` INT(11) NOT NULL,`description` TEXT NOT NULL,`created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,PRIMARY KEY (`id`),UNIQUE KEY `unique_name` (`name`),KEY `index` (`description`),KEY `fulltext_index` (`name`, `age`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;CREATE TABLE `table3` (`id` INT(11) NOT NULL AUTO_INCREMENT,`value` DECIMAL(10,2) NOT NULL,`updated_at` TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`value`, `id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;CREATE TABLE `TABLE5` (`id` INT(11) NOT NULL AUTO_INCREMENT,`value` DECIMAL(10,2) NOT NULL,`updated_at` TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

把 python 和 sql 脚本拷贝下来分别放在同一个目录下的3个文件中即可，示例在 python 3.12 环境上成功运行。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/web/26107.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

GTSAM | gtsam::ISAM2Params

GTSAM | gtsam::ISAM2Params

文章目录 概述一、定义介绍二、功能作用三、主要内容四、实例演示解释概述 本届介绍了GTSAM库的gtsam::ISAM2Params类。 一、定义介绍 gtsam::ISAM2Params 是 GTSAM 库中用于配置 ISAM2（Incremental Smoothing and Mapping 2）优化器的参数类。ISAM2 是一种用于大规模非线性优…
阅读更多...
计算机数据存储大小端模式总结

计算机数据存储大小端模式总结

计算机数据存储大小端模式总结 一&#xff0c;基本介绍1&#xff0c; 大端模式&#xff08;Big Endian&#xff09;&#xff1a;&#xff08;简记&#xff1a;高字节-低地址&#xff09;2&#xff0c;小端模式&#xff08;Little Endian&#xff09;&#xff1a;&#xff08;简…
阅读更多...
Vue2后台管理：项目开发全流程(二)

Vue2后台管理：项目开发全流程(二)

​&#x1f308;个人主页&#xff1a;前端青山 &#x1f525;系列专栏&#xff1a;vue篇 &#x1f516;人终将被年少不可得之物困其一生 依旧青山,本期给大家带来vue篇专栏内容:Vue2后台管理&#xff1a;项目开发全流程(二) 目录 功能实现 8、会员用户管理 ①使用数据模拟文…
阅读更多...
Spring IoC注解

Spring IoC注解

一、回顾反射机制 反射的调用三步&#xff1a;1&#xff09;获取类。2&#xff09;获取方法。3&#xff09;调用方法 调用方法&#xff1a;调用哪个对象&#xff0c;哪个方法&#xff0c;传什么参数&#xff0c;返回什么值。 方法&#xff08;Do&#xff09;类&#xff1a; …
阅读更多...
【QT】记录一次QT程序发布exe过程

【QT】记录一次QT程序发布exe过程

记录一次QT程序发布exe过程 使用windeploy与enigma发布独立的QT程序第一步 QT编译输出 **release** 版本第二步 QT 自带 windepoyqt 补全链接库第三步 enigma virtual box压缩打包为单一exe最后【2024-06-07 17】- 【补充】 贴一个自己用的bat脚本【**QtDeploy2exe.bat**】半自…
阅读更多...
【无标题】win10 server 服务器,安装mongodb时,遇到无法定位程序输入点BCryptHash ， 降低mongodb 版本 4.2.3

【无标题】win10 server 服务器,安装mongodb时,遇到无法定位程序输入点BCryptHash ， 降低mongodb 版本 4.2.3

安装4.2.3版本MONGODB​​​​​​​​​​a​​​​​​​​​​​​​​​​​​​​ ​​​​​​​https://fastdl.mongodb.org/win32/mongodb-win32-x86_64-2012plus-4.2.3.zip
阅读更多...
C++中的结构体——结构体嵌套结构体

C++中的结构体——结构体嵌套结构体

作用&#xff1a;结构体中的成员可以是另一个结构体 例如&#xff1a;每一个老师辅导一个学生&#xff0c;每个老师的结构体中&#xff0c;记录一个学生的结构体 示例 运行结果
阅读更多...
vue相关的2个综合案例，网页打字练习

vue相关的2个综合案例，网页打字练习

for循环的应用 /* 1. 用for循环控制台打印0到100 */ for (var i 0; i < 100; i) {console.log(i) } /* 2. 用for循环控制台打印100到0 */ for (var i 100; i > 0; i--) {console.log(i) }网页打字练习案例练习 <template><div class"main"><…
阅读更多...
[天翼杯 2021]esay_eval

[天翼杯 2021]esay_eval

[天翼杯 2021]esay_eval <?php class A{public $code "";function __call($method,$args){eval($this->code);}function __wakeup(){$this->code "";} }class B{function __destruct(){echo $this->a->a();} } if(isset($_REQUEST[poc]…
阅读更多...
MySQL的索引类型，以及各自的作用

MySQL的索引类型，以及各自的作用

MySQL的索引类型&#xff0c;以及各自的作用 常见的索引类型 主键索引&#xff08;Primary Key Index&#xff09;&#xff1a; 唯一标识表中的记录&#xff0c;确保索引列的值在整个表中是唯一的主键索引通常是唯一索引的一种特例作用&#xff1a;加速查询&#xff0c;并自动…
阅读更多...
uniapp 接口请求封装

uniapp 接口请求封装

根目录下创建 config目录 api.js request.js // request.js // 封装一个通用的网络请求函数 适当调整 function httpRequest(options) {const userToken uni.getStorageSync(access_token).token;return new Promise((resolve, reject) > {uni.request({url: ${options.ur…
阅读更多...
2-2 基于matlab的变邻域

2-2 基于matlab的变邻域

基于matlab的变邻域&#xff0c;含变惯性权重策略的自适应离散粒子群算法&#xff0c;适应函数是多式联运路径优化距离。有10城市、30城市、75城市三个案例。可直接运行。 2-2 路径规划 自适应离散粒子群算法 - 小红书 (xiaohongshu.com)
阅读更多...
新版校园跑腿外卖独立版+APP+小程序前端外卖配送平台源码（含搭建教程）

新版校园跑腿外卖独立版+APP+小程序前端外卖配送平台源码（含搭建教程）

同城校园跑腿外卖配送平台源码&#xff0c;这套目前全网还没有人分享过&#xff0c;这个是开源的&#xff0c;所以没有任何问题了&#xff0c;这套源码非常吊&#xff0c;支持自定义diy 你可以设计你的页面&#xff0c;设计你自己的风格&#xff0c;支持多校园&#xff0c;独立…
阅读更多...
打破时空界限：线上非遗文化馆如何改变非遗文化传播与保存方式？

打破时空界限：线上非遗文化馆如何改变非遗文化传播与保存方式？

一、线上非遗文化馆助力传统文化的广泛传播 1、打破时空限制&#xff0c;提升非遗文化的可达性 线上非遗文化馆利用互联网技术将非遗文化展示在虚拟平台上&#xff0c;无论身处何地&#xff0c;用户都可以通过网络访问这些资源。通过3D建模、VR等技术&#xff0c;将传统工艺、表…
阅读更多...
计算机毕业三年的我，辞职两次后找不到工作回家，此时是真的羡慕有手艺在手的人

计算机毕业三年的我，辞职两次后找不到工作回家，此时是真的羡慕有手艺在手的人

栀子花香&#xff0c;弥漫在空气中&#xff0c;却掩盖不了内心的苦涩。 半年&#xff0c;两份工作&#xff0c;两次裸辞&#xff0c;我&#xff0c;又成了一个身无分文的“废人”。 曾经&#xff0c;我也是人人羡慕的互联网人&#xff0c;月薪6K&#xff0c;过着“955”的“神…
阅读更多...
Nginx 版本升级方案

Nginx 版本升级方案

因 nginx发现漏洞、需 Nginx 的版本进行更新&#xff0c;需要用到Nginx服务器提供的平滑升级功能。 一、Nginx安装 Linux服务器 离线安装 nginx_linux 离线安装nginx 依赖包 百度云-CSDN博客 二、查看已安装的 Nginx 版本信息&#xff0c;configure 配置信息 ## nginx 目录 /…
阅读更多...
【Mac】精通或死亡Spellz Mastery or Death（角色扮演游戏)）游戏介绍

【Mac】精通或死亡Spellz Mastery or Death（角色扮演游戏)）游戏介绍

前言 今天给大家介绍一款游戏&#xff0c;《精通或死亡Spellz Mastery or Death for mac》(角色扮演游戏) 。 游戏介绍 《精通或死亡&#xff1a;Spellz Mastery or Death》是一款以魔法为核心的策略角色扮演游戏&#xff08;RPG&#xff09;&#xff0c;玩家在游戏中需要掌…
阅读更多...
uniapp 展示地图，并获取当前位置信息（精确位置）

uniapp 展示地图，并获取当前位置信息（精确位置）

使用uniapp 提供的map标签 <map :keymapIndex class"container" :latitude"latitude" :longitude"longitude" ></map> 页面初始化的时候&#xff0c;获取当前的位置信息 created() {let that thisuni.getLocation({type: gcj02…
阅读更多...
【云原生】使用kubekey部署k8s多节点及kubesphere

【云原生】使用kubekey部署k8s多节点及kubesphere

kubesphere官方部署文档 https://github.com/kubesphere/kubesphere/blob/master/README_zh.md kubuctl命令文档 https://kubernetes.io/zh-cn/docs/reference/kubectl/ k8s资源类型 https://kubernetes.io/zh-cn/docs/reference/kubectl/#%E8%B5%84%E6%BA%90%E7%B1%BB%E5%9E…
阅读更多...
linux 文件删除空间未释放问题

linux 文件删除空间未释放问题

nohup.out 文件有20G&#xff0c; 使用rm -rf nohup.out 删除后&#xff0c;磁盘空间并没有释放&#xff0c;原因是java 进程一直往 nohup.out 中写文件&#xff0c;导致nohup文件一直被占用&#xff0c;导致空间无法释放&#xff0c;要释放空间&#xff0c;就重启java服务&…
阅读更多...
最新文章

Copyright @ 2022~2023