Python数据库同步神器(一键同步)

Python数据库同步神器(一键同步)1.背景:最近公司项目的用户中心模块经常出现线上问题,测试人员需要将线上真实数据导入到测试环境中去复现Bug。公司有3套测试环境,来回切换并校验数据比较麻烦,于是就有了如下的数据库同步神器出现了。ps:鄙人代码写的比较烂,还望各位大神多指点2.界面:3.源代码:#-*-coding:utf-8-*-“””@date:2022/01/1622:00@author:Anker@python:v3.6″””importjsonimport..

大家好,又见面了,我是你们的朋友全栈君。

1.背景:

最近公司项目的用户中心模块经常出现线上问题,测试人员需要将线上真实数据导入到测试环境中去复现Bug。公司有3套测试环境,来回切换并校验数据比较麻烦,于是就有了如下的数据库同步神器。

2.界面:

Python数据库同步神器(一键同步)

3.源代码:

# -*- coding: utf-8 -*-
"""
@date:  2022/01/16 22:00
@author: Anker
@python:v3.6
"""
import json
import tkinter as tk
import tkinter.filedialog
import tkinter.messagebox
import pymysql as pymysql
from tkinter import END
from pymysql.cursors import DictCursor

# 线上数据库
online_connect = "erdshggadkjhg2wrtyuic.sfgh.aliyuncs.com"
online_user = "Anker"
online_password = "123456"
online_port = "3306"
online_db_name = "user_center"
online_table = "uuc_business_user"

# 测试数据库
test_connect = "bm-cp134dfhdh5e7uho.mysql.rds.aliyuncs.com"
test_user = "root"
test_port = "3306"
test_password = "Vr9ybte9hQJMghjkljh"
test_table = "uuc_business_user"

# 创建一个窗口
window = tk.Tk()
window.title("数据库同步工具")
window.geometry('600x600')

# 将new_user_data变量初始化
new_user_data = None


def data_synchronization():
    # 定义一个全局变量new_user_data
    global new_user_data
    # 每次触发操作之前,清空多行文本框的内容
    text.delete("1.0", "end")
    # 定义两个变量name和phone,用来获取用户在文本框输入的内容
    name = entry1.get()
    phone = entry2.get()
    # 连接线上数据库数据库
    conn_online_sql = pymysql.connect(host=online_connect, user=online_user, cursorclass=DictCursor,
                                      password=online_password, database=online_db_name)
    curs_online = conn_online_sql.cursor()
    # 通过字典去读取不同的测试环境
    db_config = radio.get()
    config = {"0": "0", "test1环境": "user_center_ding", "test2环境": "user_center_ding_test2", "test3环境": "test3"}
    test_db_name = config[db_config]

    try:
        if not (entry1.get() and entry2.get()):
            tk.messagebox.showinfo(title="提示", message="用户姓名、用户手机为必填项!")
        elif radio.get() == "0":
            tk.messagebox.showinfo(title="提示", message="请选择数据库")
        elif radio.get() == "test3环境":
            tk.messagebox.showinfo(title="提示", message="目前没有该测试数据库")
        else:
            # 连接测试数据库数据库"
            conn_test2_sql = pymysql.connect(host=test_connect, user=test_user, password=test_password,
                                             database=test_db_name)
            curs2 = conn_test2_sql.cursor()
            # 读取线上数据库的用户信息
            read_online_sql = f'''select * from `{online_db_name}`.`{online_table}` 
                        where name = "{name}" and phone = "{phone}" '''
            # 读取测试数据库的用户信息
            read_test2_sql = f'''select * from `{test_db_name}`.`{test_table}` where name = 
                        "{name}" and phone = "{phone}" '''
            # 执行sql查询,查询线上数据库,将查询出来的结果保存在online_user_data中
            curs_online.execute(read_online_sql)
            online_user_data = curs_online.fetchone()
            print("线上数据库该用户数据为:" + '\n' + str(online_user_data) + '\n')

            # 判断线上数据库是否有该用户
            if not online_user_data:
                text.insert("insert", "线上数据库没有该用户!" + '\n')
            else:
                # 执行下面这句后data为{"content": "x", "created_time": "2019-06-19 13:28:30"},是str类型
                data = json.dumps(online_user_data, default=str, ensure_ascii=False)
                # 把data转换为字典类型
                data_result = json.loads(data)
                print("转换为dict类型后的数据为:", '\n', data_result, '\n')
                # 把data再次转为json类型
                json_result = json.dumps(data_result)
                print("转换为json类型后的数据为:" + '\n' + json_result + '\n')
                # 获取返回字典data_result值的所有value的列表
                new_user_data = tuple(data_result.values())
                print(new_user_data)

            # 执行sql查询,查询测试数据库,将查询出来的结果保存在test_user_data中
            curs2.execute(read_test2_sql)
            test2_user_data = curs2.fetchone()
            # 提交数据至线上数据库
            conn_online_sql.commit()
            # 判断测试数据库是否有该用户
            if test2_user_data:
                text.insert("insert", "测试数据库已存在该用户数据!" + '\n')
            # 如果线上数据库有该用户,测试数据库没有该用户,则执行同步操作
            if online_user_data and (not test2_user_data):
                try:
                    # 插入的SQL语句
                    insert_sql = f'''insert into `{test_db_name}`.`{test_table}` VALUES {new_user_data}'''
                    # 执行SQL,向测试数据库中插入数据
                    curs2.execute(insert_sql)
                    print(insert_sql)
                    text.insert("insert", "数据同步成功!" + '\n')
                except IOError:
                    text.insert("insert", "数据同步失败!" + '\n')
            conn_test2_sql.commit()
            curs_online.close()
            curs2.close()
    except IOError:
        print("数据插入失败")
        conn_online_sql.rollback()
        tkinter.filedialog.Directory()
    conn_online_sql.close()


# 清空按钮函数:用来清空输入文本框中的内容
def clear():
    entry1.delete(0, END)  # 清空Entry文本框中的内容
    entry2.delete(0, END)  # 清空Entry文本框中的内容
    text.delete("1.0", "end")  # 清空text多行文本框中的内容
    radio.set("0")  # 清空Radiobutton标签选项框


def selection():
    text.delete("1.0", "end")
    choose = str(radio.get())
    text.insert("insert", "您选择的是:" + choose + '\n')


# 创建一个标签
tk.Label(window, text="线上用户姓名:").place(x=10, y=50)
tk.Label(window, text="线上用户手机:").place(x=10, y=120)
# 创建一个Entry文本框:用来输入一行文本字符串
entry1 = tk.Entry(window, bd=3, width=62, highlightcolor='red')
entry1.place(x=120, y=50)
entry2 = tk.Entry(window, bd=3, width=62, highlightcolor='green')
entry2.place(x=120, y=120)
# 创建一个可以输入多行文字的文本框
text = tk.Text(window, height=20, width=75, bg='white', highlightcolor='blue', highlightthickness=1)
text.place(x=35, y=240)
# 创建两个按钮:选择文件夹、退出窗口
tk.Button(window, text='开始同步', bg='grey', fg='white', command=data_synchronization).place(x=130, y=520)
tk.Button(window, text='清空', bg='grey', fg='white', command=clear).place(x=290, y=520)
tk.Button(window, text='退出', bg='grey', fg='white', command=window.quit).place(x=430, y=520)

radio = tk.StringVar()
radio.set("0")
tk.Label(window, text='请选择数据库:', bg='green', fg='white', width=12).place(x=10, y=180)
# 创建3个radiobutton选项
radio1 = tk.Radiobutton(window, text='test1环境', variable=radio, command=selection, value="test1环境")
radio1.place(x=140, y=180)
radio2 = tk.Radiobutton(window, text='test2环境', variable=radio, command=selection, value="test2环境")
radio2.place(x=290, y=180)
radio3 = tk.Radiobutton(window, text='test3环境', variable=radio, command=selection, value="test3环境")
radio3.place(x=440, y=180)

# 主循环
window.mainloop()

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/151220.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • 越狱软件源

    越狱软件源全部内容来源于网络。不对其内容的安全性负责,本人不承担任何责任。使用任何内容即表示同意此内容。软件源地址ios6五百年源apt.so/ios6vxios6分享源apt.feng/eq2wowMK越狱源apt.so/mksscc盗版暗影apt.so/anyinkeji杀手源apt.so/tanyao正版暗影apt.so/anyinkj…

  • 集合:映射,单射,满射,双射

    集合:映射,单射,满射,双射这些概念太重要了,虽然很不起眼,但是几乎所有地方都要用,一旦不清楚,会对一个问题审视错误。集合:元素汇聚在一起,就构成了一个集合。假设有集合A和集合B。映射:将A中的每一个元素,根据一个规则,对应到B中的一个元素去,比如A中是一些儿女,B中是他们的父亲,那么每一个A中的元素都可以对应到B中的唯一一个元素,这就是一种规则,即映射。因为一个人的爸爸只有一个,但是反之未必,因为一个爸爸可能有多个儿女。我们发现,这个和函数的概念非常相似,集合A相当于定义域,映射相当于f。但是集合的概念更加宽泛,正如前面所有

  • MSDN Windows 7 旗舰版 序列号

    MSDN Windows 7 旗舰版 序列号 GTWV3-KH84H-M94BG-PVT2W-JFPRW89Q2G-GBRGH-KFWRJ-Q72FD-GHYB7J3X6R-HPF6X-FQRDR-WTPM7-J92GK7KG4K-T2PTK-8YGQT-QX68X-RGRQ3TQMMV-43FFG-RGXMY-KMVFY-MB8JW8483D-TTKCX-CTDR6-XQXTH-X9JG4BVFKK-9X3FC-XPF4D-W8GTF

  • 华为模拟器 eNSP安装教程「建议收藏」

    华为模拟器 eNSP安装教程「建议收藏」随着华为网络设备的日渐普遍,华为网络设备的使用越来越多,学习华为网络路由知识的人也越来越多,华为提供的eNSP模拟软件,能够很好的为初学者提供很好的模拟学习软件。一、准备工作软件:ensp软件为免费软件,可以在华为官网上进行下载,版本:eNSPV100R002C00B320window10操作系统二、安装步骤1、下载eNSP软件:该软件为免费…

    2022年10月14日
  • laravel 学习笔记blog后台

    laravel 学习笔记blog后台

    2021年10月24日
  • 微信小程序开发入门教程

    微信小程序开发入门教程做任何程序开发要首先找到其官方文档,我们先来看看其有哪些官方文档。微信小程序开发文档链接为:https://mp.weixin.qq.com/debug/wxadoc/dev/index.html,如下图:  这里就是做微信小程序开发的全部官方文档。知道了文档的位置,下面我们来介绍下如何做一个微信小程序开发: /***************

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号