近日,某个QQ 群里的一个朋友提出一个问题,如何将一个DB 的表结构同步给另一个DB。 针对这个问题,我进行了思考与实践,具体的实现代码如下所示:
# coding:utf-8 import pymysql dbDict = {"test1":"l-beta.test1"} dbUser = "test" dbPassword = "123456" class DBUtils(): def __init__(self): self.conn = pymysql.connect(dbDict['test1'], dbUser, dbPassword) self.cursor = self.conn.cursor() def dbSelect(self, sql): print("------------------------------------") print(sql) resultList = [] self.cursor.execute(sql) result = self.cursor.fetchall() columns = self.cursor.description for val in result: tempDict = {} for cloNum in range(len(columns)): tempDict[str(columns[cloNum][0])] = val[cloNum] resultList.append(tempDict) print("---------------------打印查询结果----------------------") print(resultList) self.dbClose() return resultList def dbExcute(self, sql): print(sql) self.cursor.execute(sql) self.dbClose() def dbClose(self): self.conn.commit() self.cursor.close() self.conn.close() if __name__ == "__main__": test = DBUtils() result = test.dbSelect("select table_name from information_schema.tables where table_schema='testdb1'") for dict1 in result: test = DBUtils() create_table_sql = "create table testdb.%s as select * from testdb1.%s" % (dict1['table_name'],dict1['table_name']) print(create_table_sql) test.dbExcute(create_table_sql)示例代码操作简单,通俗易懂,所以没有过多的注释,如有疑问的小伙伴们,可在文章下方评论。
欢迎关注【无量测试之道】公众号,回复【领取资源】, Python编程学习资源干货、 Python+Appium框架APP的UI自动化、 Python+Selenium框架Web的UI自动化、 Python+Unittest框架API自动化、
资源和代码 免费送啦~ 文章下方有公众号二维码,可直接微信扫一扫关注即可。
备注:我的个人公众号已正式开通,致力于测试技术的分享,包含:大数据测试、功能测试,测试开发,API接口自动化、测试运维、UI自动化测试等,微信搜索公众号:“无量测试之道”,或扫描下方二维码:
添加关注,让我们一起共同成长!
