Als ich arbeitete, stimmte die in der Testumgebung verwendete Datenbank nicht mit der Datenbank in der Produktionsumgebung überein, als die Datenbank in unserer Testumgebung den Test abschloss und bereit war, auf die Datenbank in der Produktion aktualisiert zu werden Umgebung mussten wir ein Update-Skript vorbereiten. Wenn Sie es nicht aufschreiben, werden Sie vergessen, was geändert oder hinzugefügt wurde. Deshalb habe ich versucht, mit Python automatisch Aktualisierungsskripte zu generieren, um zu verhindern, dass mein schlechtes Gedächtnis Dinge vergisst.
Die Hauptoperationen sind wie folgt:
1. Fügen Sie die folgende Methode zur ursprünglichen basedao.py hinzu, damit die Daten der Datenbank gespeichert werden können Leicht zum Testen erhältlich. Der Vergleich zwischen der Datenbank und der Produktionsdatenbank legte den Grundstein.
1 def select_database_struts(self): 2 ''' 3 查找当前连接配置中的数据库结构以字典集合 4 ''' 5 sql = '''SELECT COLUMN_NAME, IS_NULLABLE, COLUMN_TYPE, COLUMN_KEY, COLUMN_COMMENT 6 FROM information_schema.`COLUMNS`
7 WHERE TABLE_SCHEMA="%s" AND TABLE_NAME="{0}" '''%(self.__database) 8 struts = {} 9 for k in self.__primaryKey_dict.keys():10 self.__cursor.execute(sql.format(k))11 results = self.__cursor.fetchall()12 struts[k] = {}13 for result in results:14 struts[k][result[0]] = {}15 struts[k][result[0]]["COLUMN_NAME"] = result[0]16 struts[k][result[0]]["IS_NULLABLE"] = result[1]17 struts[k][result[0]]["COLUMN_TYPE"] = result[2]18 struts[k][result[0]]["COLUMN_KEY"] = result[3]19 struts[k][result[0]]["COLUMN_COMMENT"] = result[4]20 return self.__config, strutsNach dem Login kopieren
2. Schreiben Sie ein vergleichendes Python-Skript
1 ''' 2 数据库迁移脚本, 目前支持一下几种功能： 3 1.生成旧数据库中没有的数据库表执行 SQL 脚本（支持是否带表数据）,生成的 SQL 脚本在 temp 目录下（表名.sql）。 4 2.生成添加列 SQL 脚本，生成的 SQL 脚本统一放在 temp 目录下的 depoyed.sql 中。 5 3.生成修改列属性 SQL 脚本，生成的 SQL 脚本统一放在 temp 目录下的 depoyed.sql 中。 6 4.生成删除列 SQL 脚本，生成的 SQL 脚本统一放在 temp 目录下的 depoyed.sql 中。 7 ''' 8 import json, os, sys 9 from basedao import BaseDao 10 11 temp_path = sys.path[0] + "/temp" 12 if not os.path.exists(temp_path): 13 os.mkdir(temp_path) 14 15 def main(old, new, has_data=False): 16 ''' 17 @old 旧数据库（目标数据库） 18 @new 最新的数据库（源数据库） 19 @has_data 是否生成结构+数据的sql脚本
20 ''' 21 clear_temp() # 先清理 temp 目录 22 old_config, old_struts = old 23 new_config, new_struts = new 24 for new_table, new_fields in new_struts.items(): 25 if old_struts.get(new_table) is None: 26 gc_sql(new_config["user"], new_config["password"], new_config["database"], new_table, has_data) 27 else: 28 cmp_table(old_struts[new_table], new_struts[new_table], new_table) 29 30 def cmp_table(old, new, table): 31 ''' 32 对比表结构生成 sql 33 ''' 34 old_fields = old 35 new_fields = new 36 37 sql_add_column = "ALTER TABLE `{TABLE}` ADD COLUMN `{COLUMN_NAME}` {COLUMN_TYPE} COMMENT '{COLUMN_COMMENT}';\n" 38 sql_change_column = "ALTER TABLE `{TABLE}` CHANGE `{COLUMN_NAME}` `{COLUMN_NAME}` {COLUMN_TYPE} COMMENT '{COLUMN_COMMENT}';\n" 39 sql_del_column = "ALTER TABLE `{TABLE}` DROP {COLUMN_NAME};" 40 41 if old_fields != new_fields: 42 f = open(sys.path[0] + "/temp/deploy.sql", "a", encoding="utf8") 43 content = "" 44 for new_field, new_field_dict in new_fields.items(): 45 old_filed_dict = old_fields.get(new_field) 46 if old_filed_dict is None: 47 # 生成添加列 sql 48 content += sql_add_column.format(TABLE=table, **new_field_dict) 49 else: 50 # 生成修改列 sql 51 if old_filed_dict != new_field_dict: 52 content += sql_change_column.format(TABLE=table, **new_field_dict) 53 pass 54 # 生成删除列 sql 55 for old_field, old_field_dict in old_fields.items(): 56 if new_fields.get(old_field) is None: 57 content += sql_del_column.format(TABLE=table, COLUMN_NAME=old_field) 58 59 f.write(content) 60 f.close() 61 62 def gc_sql(user, pwd, db, table, has_data): 63 ''' 64 生成 sql 文件 65 ''' 66 if has_data: 67 sys_order = "mysqldump -u%s -p%s %s %s > %s/%s.sql"%(user, pwd, db, table, temp_path, table) 68 else: 69 sys_order = "mysqldump -u%s -p%s -d %s %s > %s/%s.sql"%(user, pwd, db, table, temp_path, table) 70 os.system(sys_order) 71 72 def clear_temp(): 73 ''' 74 每次执行的时候调用这个，先清理下temp目录下面的旧文件 75 ''' 76 if os.path.exists(temp_path): 77 files = os.listdir(temp_path) 78 for file in files: 79 f = os.path.join(temp_path, file) 80 if os.path.isfile(f): 81 os.remove(f) 82 print("临时文件目录清理完成") 83 84 if __name__ == "__main__": 85 test1_config = { 86 "user" : "root",
87 "password" : "root", 88 "database" : "test1",
89 } 90 test2_config = { 91 "user" : "root",
92 "password" : "root", 93 "database" : "test2",
94 } 95 96 test1_dao = BaseDao(**test1_config) 97 test1_struts = test1_dao.select_database_struts() 98 99 test2_dao = BaseDao(**test2_config)100 test2_struts = test2_dao.select_database_struts()101 102 main(test2_struts, test1_struts)Nach dem Login kopieren
Unterstützt derzeit nur die Generierung von 4 Arten von SQL-Skripten.
Wenn Sie daran interessiert sind, gemeinsam Python zu lernen und zu diskutieren, können Sie der QQ-Gruppe beitreten: 626787819. Wenn Sie Kommentare oder Vorschläge haben, können Sie mir eine E-Mail senden: 410093793@qq.com.
Das obige ist der detaillierte Inhalt vonPython-Generierungsskript – Datenbankaktualisierung implementieren. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
