Python 實(shí)現(xiàn)數(shù)據(jù)庫(kù)(SQL)更新腳本的生成方法
我在工作的時(shí)候,在測(cè)試環(huán)境下使用的數(shù)據(jù)庫(kù)跟生產(chǎn)環(huán)境的數(shù)據(jù)庫(kù)不一致,當(dāng)我們的測(cè)試環(huán)境下的數(shù)據(jù)庫(kù)完成測(cè)試準(zhǔn)備更新到生產(chǎn)環(huán)境上的數(shù)據(jù)庫(kù)時(shí)候,需要準(zhǔn)備更新腳本,真是一不小心沒記下來(lái)就會(huì)忘了改了哪里,哪里添加了什么,這個(gè)真是非常讓人頭疼。因此我就試著用Python來(lái)實(shí)現(xiàn)自動(dòng)的生成更新腳本,以免我這爛記性,記不住事。
主要操作如下:
1.在原先 basedao.py 中添加如下方法,這樣舊能很方便的獲取數(shù)據(jù)庫(kù)的數(shù)據(jù),為測(cè)試數(shù)據(jù)庫(kù)和生產(chǎn)數(shù)據(jù)庫(kù)做對(duì)比打下了基礎(chǔ)。
def select_database_struts(self):
'''
查找當(dāng)前連接配置中的數(shù)據(jù)庫(kù)結(jié)構(gòu)以字典集合
'''
sql = '''SELECT COLUMN_NAME, IS_NULLABLE, COLUMN_TYPE, COLUMN_KEY, COLUMN_COMMENT
FROM information_schema.`COLUMNS`
WHERE TABLE_SCHEMA="%s" AND TABLE_NAME="{0}" '''%(self.__database)
struts = {}
for k in self.__primaryKey_dict.keys():
self.__cursor.execute(sql.format(k))
results = self.__cursor.fetchall()
struts[k] = {}
for result in results:
struts[k][result[0]] = {}
struts[k][result[0]]["COLUMN_NAME"] = result[0]
struts[k][result[0]]["IS_NULLABLE"] = result[1]
struts[k][result[0]]["COLUMN_TYPE"] = result[2]
struts[k][result[0]]["COLUMN_KEY"] = result[3]
struts[k][result[0]]["COLUMN_COMMENT"] = result[4]
return self.__config, struts
2.編寫對(duì)比的Python腳本
'''
數(shù)據(jù)庫(kù)遷移腳本, 目前支持一下幾種功能:
1.生成舊數(shù)據(jù)庫(kù)中沒有的數(shù)據(jù)庫(kù)表執(zhí)行 SQL 腳本(支持是否帶表數(shù)據(jù)),生成的 SQL 腳本在 temp 目錄下(表名.sql)。
2.生成添加列 SQL 腳本,生成的 SQL 腳本統(tǒng)一放在 temp 目錄下的 depoyed.sql 中。
3.生成修改列屬性 SQL 腳本,生成的 SQL 腳本統(tǒng)一放在 temp 目錄下的 depoyed.sql 中。
4.生成刪除列 SQL 腳本,生成的 SQL 腳本統(tǒng)一放在 temp 目錄下的 depoyed.sql 中。
'''
import json, os, sys
from basedao import BaseDao
temp_path = sys.path[0] + "/temp"
if not os.path.exists(temp_path):
os.mkdir(temp_path)
def main(old, new, has_data=False):
'''
@old 舊數(shù)據(jù)庫(kù)(目標(biāo)數(shù)據(jù)庫(kù))
@new 最新的數(shù)據(jù)庫(kù)(源數(shù)據(jù)庫(kù))
@has_data 是否生成結(jié)構(gòu)+數(shù)據(jù)的sql腳本
'''
clear_temp() # 先清理 temp 目錄
old_config, old_struts = old
new_config, new_struts = new
for new_table, new_fields in new_struts.items():
if old_struts.get(new_table) is None:
gc_sql(new_config["user"], new_config["password"], new_config["database"], new_table, has_data)
else:
cmp_table(old_struts[new_table], new_struts[new_table], new_table)
def cmp_table(old, new, table):
'''
對(duì)比表結(jié)構(gòu)生成 sql
'''
old_fields = old
new_fields = new
sql_add_column = "ALTER TABLE `{TABLE}` ADD COLUMN `{COLUMN_NAME}` {COLUMN_TYPE} COMMENT '{COLUMN_COMMENT}';\n"
sql_change_column = "ALTER TABLE `{TABLE}` CHANGE `{COLUMN_NAME}` `{COLUMN_NAME}` {COLUMN_TYPE} COMMENT '{COLUMN_COMMENT}';\n"
sql_del_column = "ALTER TABLE `{TABLE}` DROP {COLUMN_NAME};"
if old_fields != new_fields:
f = open(sys.path[0] + "/temp/deploy.sql", "a", encoding="utf8")
content = ""
for new_field, new_field_dict in new_fields.items():
old_filed_dict = old_fields.get(new_field)
if old_filed_dict is None:
# 生成添加列 sql
content += sql_add_column.format(TABLE=table, **new_field_dict)
else:
# 生成修改列 sql
if old_filed_dict != new_field_dict:
content += sql_change_column.format(TABLE=table, **new_field_dict)
pass
# 生成刪除列 sql
for old_field, old_field_dict in old_fields.items():
if new_fields.get(old_field) is None:
content += sql_del_column.format(TABLE=table, COLUMN_NAME=old_field)
f.write(content)
f.close()
def gc_sql(user, pwd, db, table, has_data):
'''
生成 sql 文件
'''
if has_data:
sys_order = "mysqldump -u%s -p%s %s %s > %s/%s.sql"%(user, pwd, db, table, temp_path, table)
else:
sys_order = "mysqldump -u%s -p%s -d %s %s > %s/%s.sql"%(user, pwd, db, table, temp_path, table)
os.system(sys_order)
def clear_temp():
'''
每次執(zhí)行的時(shí)候調(diào)用這個(gè),先清理下temp目錄下面的舊文件
'''
if os.path.exists(temp_path):
files = os.listdir(temp_path)
for file in files:
f = os.path.join(temp_path, file)
if os.path.isfile(f):
os.remove(f)
print("臨時(shí)文件目錄清理完成")
if __name__ == "__main__":
test1_config = {
"user" : "root",
"password" : "root",
"database" : "test1",
}
test2_config = {
"user" : "root",
"password" : "root",
"database" : "test2",
}
test1_dao = BaseDao(**test1_config)
test1_struts = test1_dao.select_database_struts()
test2_dao = BaseDao(**test2_config)
test2_struts = test2_dao.select_database_struts()
main(test2_struts, test1_struts)
目前只支持了4種SQL腳本的生成。
總結(jié)
以上所述是小編給大家介紹的Python 實(shí)現(xiàn)數(shù)據(jù)庫(kù)(SQL)更新腳本的生成方法,希望對(duì)大家有所幫助,如果大家有任何疑問(wèn)歡迎給我留言,小編會(huì)及時(shí)回復(fù)大家的!
相關(guān)文章
Python中對(duì)對(duì)象列表進(jìn)行排序的多種方式實(shí)現(xiàn)
在軟件開發(fā)中,我們經(jīng)常需要對(duì)數(shù)據(jù)進(jìn)行排序以滿足特定的需求,在 Python 中,排序是一個(gè)常見的操作,可以通過(guò)多種方式實(shí)現(xiàn),需要的朋友可以參考下2025-01-01
django第一個(gè)項(xiàng)目127.0.0.1:8000不能訪問(wèn)的解決方案詳析
django項(xiàng)目服務(wù)啟動(dòng)后無(wú)法通過(guò)127.0.0.1訪問(wèn),下面這篇文章主要給大家介紹了關(guān)于django第一個(gè)項(xiàng)目127.0.0.1:8000不能訪問(wèn)的解決方案,需要的朋友可以參考下2022-10-10
centos6.8安裝python3.7無(wú)法import _ssl的解決方法
這篇文章主要介紹了centos6.8安裝python3.7無(wú)法import _ssl的解決方法,小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2018-09-09
對(duì)pandas中apply函數(shù)的用法詳解
下面小編就為大家分享一篇對(duì)pandas中apply函數(shù)的用法詳解,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2018-04-04

