手把手教你如何使用 Python 操作 Mysql 进行数据库的 diff

Wu_Candy

发布于 2022-07-04 12:39:47

97000

代码可运行

文章被收录于专栏：无量测试之道无量测试之道

运行总次数：0

代码可运行

这是无量测试之道的第193篇原创

分享主题：如何使用 Python 操作 Mysql 实现不同环境相同库的 diff

一、适用场景

项目工作中，我们会遇到测试环境特别多的情况，例如：n套beta环境，多套预生产环境，多套uat环境等各种测试环境。为保证所有测试环境表结构上的一致性，如果只是单纯地靠人工来检查并更新，未免显得太过吃力且效率低下，还容易在实施的过程中出现遗漏的问题。

因此使用自动化脚本来完成这项工作就显得格外重要了，今天分享的主要内容就是通过自动化脚本协助你找到不同测试环境之间差异化的内容，进而可以避免同步过程中出现的遗漏问题。

二、代码实现

1、前置处理与基础配置

import pymysql
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import smtplib
import time

#忽略掉的db字典
ignoreDb={'information_schema': 'information_schema',
          'mysql': 'mysql',
          'performance_schema': 'performance_schema',
          'sys': 'sys'
          }

#忽略掉的db在查询sql中使用的
ignoreDbSQL="('information_schema',  'mq_store','performance_schema',  'sys','edsystem')"

#收diff报告的邮箱地址
emails='tony.wu@test-la.com'

#排除不在diff范围内的表名称列表
exclude_table = ['tt', 'hurdle_policy_back', 'preferences_0309', 'preferences_0524', 'preferences_0310','t1','t2','mock_data']

2、用来发送diff报告

def send_mail(receivers, title, content):
    sender = 'qa.notice@test-la.com'
    mailto = receivers.split(",")
    try:
        msg = MIMEMultipart()
        msg['Subject'] = title
        to_user = ",".join(mailto)

        print("receivers...", to_user)
        msg['to'] = to_user
        msg['From'] = sender

        body = MIMEText(content, _subtype='html', _charset='utf-8')
        msg.attach(body)
        smtp = smtplib.SMTP('smtp.office365.com', 587)
        smtp.starttls()
        print("sending")
        smtp.login("qa.notice@test-la.com", "test123456")
        smtp.sendmail(sender, mailto, msg.as_string())
        print("send")
        smtp.quit()
    except smtplib.SMTPException as e:
        print(e)

3、查询获取所有的列名数据

def queryAllColumns(mycursor):
    sql = " SELECT TABLE_SCHEMA 库名,TABLE_NAME 表名,COLUMN_NAME 列名, COLUMN_TYPE 数据类型, DATA_TYPE 字段类型, " \
          "CHARACTER_MAXIMUM_LENGTH 长度, IS_NULLABLE 是否为空, COLUMN_DEFAULT 默认值, " \
          "COLUMN_COMMENT 备注  FROM INFORMATION_SCHEMA.COLUMNS " \
          "where table_schema not in " + ignoreDbSQL;
    # print(sql)
    mycursor.execute(sql)
    result = mycursor.fetchall()

    d = {};
    for x in result:
        r={}
        r['TABLE_SCHEMA'] = x[0];
        r['TABLE_NAME'] = x[1];
        if(x[1] in exclude_table):
            continue
        r['COLUMN_NAME'] = x[2];
        r['COLUMN_TYPE'] = x[3];
        r['DATA_TYPE'] = x[4];
        r['CHARACTER_MAXIMUM_LENGTH'] = x[5];
        r['IS_NULLABLE'] = x[6];
        r['COLUMN_DEFAULT'] = x[7];
        r['COLUMN_COMMENT'] = x[8];
        d[x[0] + "." + x[1] + "." + x[2]] = r;

4、最终d的数据模式如下并将其返回

    d=
    {'route_config_meta.entity_sharding_config.create_date': 
      {'TABLE_SCHEMA': 'route_config_meta', 'TABLE_NAME': 'entity_sharding_config', 'COLUMN_NAME': 'create_date',
     'COLUMN_TYPE': 'datetime', 'DATA_TYPE': 'datetime', 'CHARACTER_MAXIMUM_LENGTH': None, 'IS_NULLABLE': 'NO', 
     'COLUMN_DEFAULT': None, 'COLUMN_COMMENT': '创建时间'}, 
     'route_config_meta.entity_sharding_config.update_time': 
      {'TABLE_SCHEMA': 'route_config_meta', 'TABLE_NAME': 'entity_sharding_config', 'COLUMN_NAME': 'update_time', 
      'COLUMN_TYPE': 'datetime', 'DATA_TYPE': 'datetime', 'CHARACTER_MAXIMUM_LENGTH': None, 'IS_NULLABLE': 'NO',
       'COLUMN_DEFAULT': 'CURRENT_TIMESTAMP', 'COLUMN_COMMENT': '更新时间'}
    }
    return d;

5、查询获取所有的索引数据

def queryAllIndex(mycursor):
    sql = "Select TABLE_SCHEMA 库名称,TABLE_NAME 表名称,INDEX_NAME 索引的名称,SEQ_IN_INDEX 索引中的列序列号,COLUMN_NAME 列名称 from INFORMATION_SCHEMA.STATISTICS " \
          "where TABLE_SCHEMA not in " + ignoreDbSQL;

    sql=
    Select TABLE_SCHEMA,TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX,COLUMN_NAME from INFORMATION_SCHEMA.STATISTICS where TABLE_SCHEMA 
    not in ('information_schema',  'mq_store',  'mysql',  'performance_schema',   'slow_query_log','repeater',
    'repeater_console','sys','edsystem','crs_adapter','oxi-adapter')

    mycursor.execute(sql)

    result = mycursor.fetchall()
    # print(result)

    result=
    (
     ('auth', 'authorities', 'PRIMARY', 1, 'id'), 
     ('auth', 'authorities', 'idx_code', 1, 'code')
    )

    d = {};
    for x in result:
        r = {}
        r['TABLE_SCHEMA'] = x[0];
        r['TABLE_NAME'] = x[1];
        if (x[1] in exclude_table):
            continue
        r['INDEX_NAME'] = x[2];
        r['SEQ_IN_INDEX'] = x[3];
        r['COLUMN_NAME'] = x[4];
        d[x[0]+"."+x[1]+"."+x[2]]=r;

    d=
    {
    'auth.authorities.PRIMARY': {'TABLE_SCHEMA': 'auth', 'TABLE_NAME': 'authorities', 'INDEX_NAME': 'PRIMARY', 'SEQ_IN_INDEX': 1, 'COLUMN_NAME': 'id'}, 
    'auth.authorities.idx_code': {'TABLE_SCHEMA': 'auth', 'TABLE_NAME': 'authorities', 'INDEX_NAME': 'idx_code', 'SEQ_IN_INDEX': 1, 'COLUMN_NAME': 'code'}
    }
    return d;

6、查询获取数据库实例名称

def queryDbs(mycursor):
    sql = 'show databases';
    mycursor.execute(sql)
    result = mycursor.fetchall()

    d = {};
    for x in result:
        d[x[0]] = x[0];
    return d;

7、构建成html格式的diff报表出来

def buildHtml(db1,db2,cst,ist,tip):
    # tip = db2.get('name') + "("  + db2.get('host') + ")"+ " 对比 " + db1.get('name') + "(" + db1.get('host') + ")"
    str = '<!DOCTYPE html> <html> <meta charset="utf-8"> <head> <style type="text/css"> table.gridtable {font-family: verdana,arial,sans-serif; font-size:11px; color:#333333; border-width: 1px; border-color: #666666; border-collapse: collapse; } table.gridtable th {border-width: 1px; padding: 8px; border-style: solid; border-color: #666666; background-color: #dedede; } table.gridtable td {border-width: 1px; padding: 8px; border-style: solid; border-color: #666666; background-color: #ffffff; } </style> </head>';
    str = str+'<body>'

    str = str+' <table class="gridtable">'
    str = str+ tip+'  缺少字段：'
    str = str+' <tr> <th>database</th> <th>table</th> <th>column</th> <tr/> '

    for x in cst:
        xs = x.split('.')
        str=str + '<tr> <td>'+xs[0]+'</td> <td>'+xs[1]+'</td> <td>'+xs[2]+'</td> <tr/>'
    str = str + ' </table>'


    str = str + ' <table class="gridtable">'
    str = str + tip + '  缺少索引：'
    str = str + ' <tr> <th>database</th> <th>table</th> <th>index</th> <tr/> <tr>'

    for x in ist:
        xs = x.split('.')
        str = str + '<tr> <td>' + xs[0] + '</td> <td>' + xs[1] + '</td> <td>' + xs[2] + '</td> <tr/>'

    str = str + ' </table>'
    str = str+'</body> </html>';
    return str;

8、数据库的字段与索引diff 对比，并生成diff报告，发送邮件

def diff(db1,db2):
    mydb1 = pymysql.connect(
        host=db1.get('host'),
        user=db1.get('user'),
        passwd=db1.get('password')
    );

    mydb2 = pymysql.connect(
        host=db2.get('host'),
        user=db2.get('user'),
        passwd=db2.get('password')
    )

    mycursor1 = mydb1.cursor()
    mycursor2 = mydb2.cursor()

    #获取两个库里面的所有字段相关值
    all_columns1 = queryAllColumns(mycursor1)
    all_columns2 = queryAllColumns(mycursor2)

    #获取两个库里面的所有索引相关值
    all_index1 = queryAllIndex(mycursor1)
    all_index2 = queryAllIndex(mycursor2)

    mycursor1.close()
    mycursor2.close()
    mydb1.close()
    mydb2.close()

    #定义了一个提示信息标题头
    tip = db2.get('name') +" 对比 " + db1.get('name') + "--("  + db2.get('host') + " 对比 "+ db1.get('host') + ")"

    all_columns1的数据格式与如下all_index1雷同（{key:value}），但是数据值上是有差异的。

    cs1 = set(all_columns1)
    cs2 = set(all_columns2)
    cst = cs1.difference(cs2)

9、索引的示例数据如下

    all_index1=
    {
    'auth.authorities.PRIMARY': {'TABLE_SCHEMA': 'auth', 'TABLE_NAME': 'authorities', 'INDEX_NAME': 'PRIMARY', 'SEQ_IN_INDEX': 1, 'COLUMN_NAME': 'id'}, 
    'auth.authorities.idx_code': {'TABLE_SCHEMA': 'auth', 'TABLE_NAME': 'authorities', 'INDEX_NAME': 'idx_code', 'SEQ_IN_INDEX': 1, 'COLUMN_NAME': 'code'}
    }

10、使用set做去重处理,只留下唯一的key

    is1 = set(all_index1)
    is2 = set(all_index2)
    #
    ist = is1.difference(is2)

    ist=
    {'rate.tt.PRIMARY', 'entity_storage_1.reservation_log_0.transaction_id', 'rate.rate_header.idx_tcode_tax_include'}

    content = buildHtml(db1,db2,sorted(cst),sorted(ist),tip)

    fmt = '%Y-%m-%d %a %H:%M:%S'  # 定义时间显示格式
    nowtime = time.strftime(fmt, time.localtime(time.time()))  # 把传入的元组按照格式，输出字符串
    print ('当前的时间：', nowtime)

    if len(ist) >= 0 or len(cst) >0:
        send_mail(emails,'DBDIFF:'+tip ,content)

if __name__ == '__main__':
    beta1 = {'name': 'beta1', 'host': '10.7.36.34', 'user': 'root', 'password': '123456'}
    beta2 = {'name': 'beta2', 'host': '10.7.36.2', 'user': 'root', 'password': '123456'}

    diff(beta1, beta2)

11、输出的邮件内容如下

beta2 对比 beta1--(10.7.36.2 对比 10.7.36.34) 缺少字段： 
database table column
db1 task_statistic create_date
db1 task_statistic execute_task_count

beta2 对比 beta1--(10.7.36.2 对比 10.7.36.34) 缺少索引： 
database table index
db1 award_test idx_membership_related_id
db2 record_flow idx_related_member_id