Python 实现数据库更新脚本的生成方法

yipeiwu_com5年前Python基础

我在工作的时候,在测试环境下使用的数据库跟生产环境的数据库不一致,当我们的测试环境下的数据库完成测试准备更新到生产环境上的数据库时候,需要准备更新脚本,真是一不小心没记下来就会忘了改了哪里,哪里添加了什么,这个真是非常让人头疼。因此我就试着用Python来实现自动的生成更新脚本,以免我这烂记性,记不住事。

主要操作如下:

1.在原先 basedao.py 中添加如下方法,这样旧能很方便的获取数据库的数据,为测试数据库和生产数据库做对比打下了基础。

def select_database_struts(self):
    '''
    查找当前连接配置中的数据库结构以字典集合
    '''
    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.编写对比的Python脚本

'''
数据库迁移脚本, 目前支持一下几种功能:
1.生成旧数据库中没有的数据库表执行 SQL 脚本(支持是否带表数据),生成的 SQL 脚本在 temp 目录下(表名.sql)。
2.生成添加列 SQL 脚本,生成的 SQL 脚本统一放在 temp 目录下的 depoyed.sql 中。
3.生成修改列属性 SQL 脚本,生成的 SQL 脚本统一放在 temp 目录下的 depoyed.sql 中。
4.生成删除列 SQL 脚本,生成的 SQL 脚本统一放在 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 旧数据库(目标数据库)
  @new 最新的数据库(源数据库)
  @has_data 是否生成结构+数据的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):
  '''
  对比表结构生成 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():
  '''
  每次执行的时候调用这个,先清理下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("临时文件目录清理完成")

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脚本的生成。

以上这篇Python 实现数据库更新脚本的生成方法就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持【听图阁-专注于Python设计】。

相关文章

python wxpython 实现界面跳转功能

python wxpython 实现界面跳转功能

用wxpython设计界面时可能会出现界面嵌套的情况 这样就需要进行界面的跳转 但是貌似wxpython没提供界面跳转的方式(也可能是我菜。。。) 所以就需要借助threading模块...

python绘图库Matplotlib的安装

python绘图库Matplotlib的安装

本文简单介绍了Python绘图库Matplotlib的安装,简介如下: matplotlib是python最著名的绘图库,它提供了一整套和matlab相似的命令API,十分适合交互式地...

Python一键安装全部依赖包的方法

requirements.txt用来记录项目所有的依赖包和版本号,只需要一个简单的pip命令就能完成。 pip freeze >requirements.txt 然后就可以用...

PyQt5实现拖放功能

PyQt5实现拖放功能

在这节教程中,我们将探讨PyQt5中的拖放操作。 在计算机图形用户界面(GUI)中,拖放是在某个虚拟对象上点击并拖动到另一个位置或虚拟对象上的操作。它通常用于调用多个动作,或为两个抽象对...

python环形单链表的约瑟夫问题详解

题目: 一个环形单链表,从头结点开始向后,指针每移动一个结点,就计数加1,当数到第m个节点时,就把该结点删除,然后继续从下一个节点开始从1计数,循环往复,直到环形单链表中只剩下了一个...