最近寫了一個工具(比較兩套測試環境mysql數據庫中表、表字段、索引的差異,基於python)通過文章簡單介紹下工具的相關內容
- 工具名稱
- 主要功能
- 具體使用方法
- 部分實現代碼
- 後續
一、工具名稱:
dbstructsync (python庫)
二、主要功能:
比較兩套環境中mysql指定庫中表、表字段及索引的差異,返回同步的sql ,裡面包含建表,修改索引,修改字段的sql .
A環境的數據庫db 作為sourcedb, B環境的數據庫db targetdb ,程序邏輯比較的是 sourcedb 與targetdb 的差異,然後返回一個list的數據類型
list中包含新建表sql,修改、增加字段sql, 刪除、新增索引sql
現在總共有3個版本,0.0.1 和0.0.2 存在一定的bug, 所以請使用最新的0.0.3版本
其他說明:由於是剛完成不久的程序,所以暫時不對最終結果sql進行執行,避免對使用過程中產生不好的影響,這個版本大家可以通過python 自行選擇需要執行哪些操作;隨着之後程序的逐步深入修改和演變,會將執行sql這一步也都加進去
同時也會優化使用方式,讓使用這個工具的小夥伴更容易操作
三、具體使用方法:
1、 pip install -i https://pypi.python.org/pypi dbstructsync
在代碼里引入使用,
from DbStructSync import cli
result=cli.db_sync(sourcedb, targetdb)
#sourcedb,targetdb是兩個dict的參數,具體參數看下面
# 這裏得到的 result = ['use 庫;',
# 'CREATE TABLE `test_async` (\n `test_async` #varchar(30) NOT NULL,\n `aa` varchar(400) DEFAULT NULL,\n #PRIMARY KEY (`test_async`)\n) ENGINE=InnoDB DEFAULT #CHARSET=utf8;',
# 'drop index `index_chaxx` on chanxx_auto_puxx_conf;',
# 'create index `index_chaxx` on #chanxx_auto_puxx_conf(`channel_nxx`,`channel_prxx`) USING #BTREE;']
#result 中包含 use 庫;
# 如果有少的表,會有 create table的數據; 如果有不同的索引,會#存在drop index 和create index的sql;
# 如果有不同的字段,會有alter table的sql ;
#只需要對這個結果,再通過pymysql的一些數據庫操作就可以保證 sourcedb #的內容與taragetdb一致。
2、同時還支持命令行操作,代碼寫入到x.py代碼中
result = cli.db_sync_commandline()
python x.py --source host=10.1.1.x,port=3306,user=root,passwd=root,db=investx --target host=10.1.1.x,port=3306,user=root,passwd=root,db=investx
命令行中 –source key=value;key2=value2;key3=value3 –target key=value;key2=value2;key3=value3
–source, –target 是兩給必輸的參數,後續的值會作為一個dict類型傳入程序。 –source是源庫的信息, –target是目標庫的信息
還包括其他幾個命令參數 –only-index , –only-fields ; –only-index 只比較索引差異, –only-fields 只比較字段差異, 非必填,默認都為False
四、部分實現代碼:
def diff_tables(sourcetable, targettable):
'''
:param sourcetable: 源數據庫的表名列表
:param targettable: 目的數據庫的表名列表
:return: 返回dict,包含三種結果,源庫多的表,目標庫多的表,相同的表
'''
logger.info('開始比較兩個庫中表的差異,源庫表{},目標庫表{}'.format(sourcetable, targettable))
table_result={}
if not isinstance(sourcetable, list) or not isinstance(targettable, list):
raise TypeError('sourcetable , targettable的類型不是list')
source_diff = set(sourcetable) - set(targettable)
target_diff = set(targettable) - set(sourcetable)
same_tables = set(sourcetable)& set(targettable)
table_result['source'] = source_diff
table_result['target'] = target_diff
table_result['same'] = same_tables
logger.info('兩個庫中表的差異結果{}'.format(table_result))
return table_result
def diff_indexs_fields(sourcesql, targetsql, type=1):
'''
:param sourcesql: 源數據庫表的創建sql
:param targetsql: 目標數據表的創建sql
:return: 記錄下索引不一致的地方
'''
result = {}
logger.info('解析語句中的索引字段,並進行比較索引')
sourcesql = parse_str(sourcesql) # 從括號中提取需要的內容
#logger.info('從括號中提取出來的信息數據{}'.format(sourcesql))
sourcesql = lists2str(sourcesql) #將list轉換為str,並對數據的空格數據進行處理
logger.info('解析完的數據的信息{}'.format(sourcesql))
sourcesql = sourcesql.split('\n') #將str按照'\\n'進行分割
logger.info('解析完數據之後的信息{}'.format(sourcesql))
targetsql = parse_str(targetsql)
targetsql = lists2str(targetsql)
targetsql = targetsql.split('\n')
if type ==1:
source_index = parse_fields(sourcesql,type)
target_index = parse_fields(targetsql,type)
result= compare_indexs_field(source_index, target_index, type)
elif type ==2:
source_field_sql = parse_fields(sourcesql, type)
target_field_sql = parse_fields(targetsql, type)
result = compare_indexs_field(source_field_sql, target_field_sql, type)
return result
def dict2sql(dict_str):
'''
將解析完成的數據轉換為對應的可執行sql
:param dict_str:
:return:
'''
dict_str = copy.deepcopy(dict_str) # 做一個深度copy,可以放心的進行相關數據處理
if not isinstance(dict_str, dict):
raise TypeError('調用方法{}參數不是dict類型,請確認'.format('dict2sql'))
#獲取db名字
for key ,value in dict_str.items():
dbname = key
logger.info('數據庫名{}'.format(dbname))
for table, table_desc in value.get('source').items():
if table =='create_table':
#create_table_sql = lists2str(table_desc)
dict_str[dbname]['source'][table] = table_desc
#其他的都是table的名字
logger.info('數據庫的修改語句:{}'.format(table_desc))
else:
logger.info('對於索引和字段的解析原始數據{}'.format(table_desc))
if table_desc.get('index'):
create_index_sql_lists=[]
#create_index_sql_lists.append('use {};'.format(dbname))
index_lists= (table_desc.get('index'))
result_index= parse_comma_split(str(index_lists)[1:-1])
for i in result_index:
if i.strip().startswith('\'KEY'):
#print(i.strip())
index_values = parse_space_split(i.strip())
drop_index_sql= 'drop index {} on {}'.format(index_values[1],table )
if len(index_values)<=3:
create_index_sql='create index {} on {}{} '.format(index_values[1], table, index_values[2])
else:
create_index_sql='create index {} on {}{} {}'.format(index_values[1], table, index_values[2], ' '.join(index_values[3:]))
create_index_sql_lists.append(drop_index_sql)
create_index_sql_lists.append(create_index_sql)
if i.strip().startswith('\'UNIQUE KEY'):
index_values = parse_space_split(i.strip())
drop_index_sql = 'drop index {} on {}'.format(index_values[2], table)
if len(index_values) <= 4:
create_index_sql = 'create unique index {} on {}{} '.format(index_values[2], table,
index_values[3])
else:
create_index_sql = 'create unique index {} on {}{} {}'.format(index_values[2], table,
index_values[3],
' '.join(index_values[4:]),
)
create_index_sql_lists.append(drop_index_sql)
create_index_sql_lists.append(create_index_sql)
logger.info('表{}解析出來的索引的修改sql{}'.format(table, create_index_sql_lists))
dict_str[dbname]['source'][table]['index'] = create_index_sql_lists
if table_desc.get('fields'):
create_fields_sql_lists=[]
#create_fields_sql_lists.append('use {};'.format(dbname))
modify_field_sqls = table_desc.get('fields').get('modify',None)
create_field_sqls=table_desc.get('fields').get('lose',None)
if modify_field_sqls:
for modify_field_sql in modify_field_sqls:
sql_indexs = parse_space_split(str(modify_field_sql)[0:-1])
#print(sql_indexs)
alter_fields_sql='alter table {} modify column {} {} {}'.format(table, sql_indexs[0],sql_indexs[1],' '.join(sql_indexs[2:]))
create_fields_sql_lists.append(alter_fields_sql)
if create_field_sqls:
for create_field_sql in create_field_sqls:
sql_indexs = parse_space_split(str(create_field_sql)[0:-1])
create_fields_sql='alter table {} add column {} {}'.format(table, sql_indexs[0],' '.join(sql_indexs[2:]))
create_fields_sql_lists.append(create_fields_sql)
logger.info('表{}解析出來的字段的修改sql{}'.format(table,create_fields_sql_lists))
dict_str[dbname]['source'][table]['fields'] = create_fields_sql_lists
return dict_str # 返回給一個全部是可執行sql的dict
五、後續:
1、對使用過程中遇到對bug進行修復
2、對代碼進行優化
3、增加其他相關功能,讓工具越來越好用
4、希望使用的小夥伴多提意見,未來成為一個好用的小工具
本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】
※網頁設計公司推薦更多不同的設計風格,搶佔消費者視覺第一線
※廣告預算用在刀口上,網站設計公司幫您達到更多曝光效益
※自行創業 缺乏曝光? 下一步"網站設計"幫您第一時間規劃公司的門面形象