Python如何查看兩個(gè)數(shù)據(jù)庫的同名表的字段名差異
更新時(shí)間：2022年05月10日 12:05:19 作者：XerCis
這篇文章主要介紹了Python如何查看兩個(gè)數(shù)據(jù)庫的同名表的字段名差異，具有很好的參考價(jià)值，希望對大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教
查看兩個(gè)數(shù)據(jù)庫的同名表的字段名差異
問題描述
開發(fā)過程中有多個(gè)測試環(huán)境，測試環(huán)境 A 加了字段，測試環(huán)境 B 忘了加，字段名對不上，同一項(xiàng)目就報(bào)錯(cuò)了
CREATE DATABASE `a` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci'; CREATE DATABASE `b` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci'; USE `a`; CREATE TABLE `student` ( `id` int(11) AUTO_INCREMENT, `class_id` int(11) DEFAULT NULL, `name` varchar(255), `birthday` date, `chinese` int(11), `math` int(11), `english` int(11), PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8; USE `b`; CREATE TABLE `student` ( `id` int(11) NOT NULL AUTO_INCREMENT, `class_id` int(11) DEFAULT NULL, `name` varchar(255) DEFAULT NULL, `birthday` date DEFAULT NULL, `gender` tinyint(4) DEFAULT NULL, `chinese` int(11) DEFAULT NULL, `math` int(11) DEFAULT NULL, `english` int(11) DEFAULT NULL, `physics` int(11) DEFAULT NULL, `chemistry` int(11) DEFAULT NULL, `biology` int(11) DEFAULT NULL, PRIMARY KEY (`id`), KEY `idx_class_id` (`class_id`) USING BTREE ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
解決方案
安裝
pip install SQLAlchemy pip install pymysql
代碼
from collections import defaultdict
import sqlalchemy
from sqlalchemy.engine.reflection import Inspector
def get_table_column_map(inspector: Inspector):
"""獲取數(shù)據(jù)庫中所有表對應(yīng)的字段"""
table_column_map = defaultdict(set)
table_names = inspector.get_table_names()
for table_name in table_names:
columns = inspector.get_columns(table_name) # 字段信息
# indexes = inspect.get_indexes(table_name) # 索引信息
for column in columns:
table_column_map[table_name].add(column['name'])
return table_column_map
def compare_table_column_difference(a, b):
"""對比同名表字段的差異"""
table_names = set(a) & set(b) # 同名表
for table_name in table_names:
columns_a = a[table_name]
columns_b = b[table_name]
difference = columns_a ^ columns_b
if difference:
print(table_name, difference)
engine = sqlalchemy.create_engine('mysql+pymysql://root:123456@127.0.0.1:3306/a')
engine1 = sqlalchemy.create_engine('mysql+pymysql://root:123456@127.0.0.1:3306/b')
inspector = sqlalchemy.inspect(engine)
inspector1 = sqlalchemy.inspect(engine1)
table_column_map = get_table_column_map(inspector)
table_column_map1 = get_table_column_map(inspector1)
compare_table_column_difference(table_column_map, table_column_map1)
# student {'gender', 'chemistry', 'physics', 'biology'}
mysql-utilities
也可以使用專門的工具——mysql-utilities 中的 mysqldiff
文檔
mysqldiff --help
命令
mysqldiff --server1=user:pass@host:port --server2=user:pass@host:port db1.object1:db2.object1
例子
mysqldiff --server1=root:123456@127.0.0.1:3306 --server2=root:123456@127.0.0.1:3306 a.student:b.student
效果
--- a.student +++ b.student @@ -3,8 +3,13 @@ `class_id` int(11) DEFAULT NULL, `name` varchar(255) DEFAULT NULL, `birthday` date DEFAULT NULL, + `gender` tinyint(4) DEFAULT NULL, `chinese` int(11) DEFAULT NULL, `math` int(11) DEFAULT NULL, `english` int(11) DEFAULT NULL, - PRIMARY KEY (`id`) + `physics` int(11) DEFAULT NULL, + `chemistry` int(11) DEFAULT NULL, + `biology` int(11) DEFAULT NULL, + PRIMARY KEY (`id`), + KEY `idx_class_id` (`class_id`) USING BTREE ) ENGINE=InnoDB DEFAULT CHARSET=utf8
優(yōu)點(diǎn)：比較內(nèi)容更詳盡
缺點(diǎn)：無法批量，要自己指定比較對象（稍微改寫即可克服）
Python數(shù)據(jù)庫之間差異對比
此腳本用于兩個(gè)數(shù)據(jù)庫之間的表、列、欄位、索引的差異對比。
cat oracle_diff.py
#!/home/dba/.pyenv/versions/3.5.2/bin/python
#coding=utf-8
import cx_Oracle
import time
import difflib
import os
v_host=os.popen('echo $HOSTNAME')
class Oracle_Status_Output():
? ? def __init__(self,username,password,tns):
? ? ? ? try:
? ? ? ? ? ? self.db = cx_Oracle.connect(username,password,tns)
? ? ? ? ? ? self.cursor = self.db.cursor()
? ? ? ? except Exception as e:
? ? ? ? ? ? print('Wrong')
? ? ? ? ? ? print(e)
? ? def schemas_tables_count(self,sql,db):
? ? ? ? try: ?
? ? ? ? ? ? self.cursor.execute(sql)
? ? ? ? ? ? v_result=self.cursor.fetchall()
? ? ? ? ? ? #print(v_result)
? ? ? ? ? ? count = 0
? ? ? ? ? ? for i in range(len(v_result)):
? ? ? ? ? ? ? ? #print(v_result[i][1],'--',v_result[i][0])
? ? ? ? ? ? ? ? count = int(v_result[i][0]) + count
? ? ? ? ? ? print(db,'Count Tables','--',count)
? ? ? ? except Exception as e:
? ? ? ? ? ? print('Wrong--schemas_tables_count()')
? ? ? ? ? ? print(e)
? ? def schemas_tables_list(self,sql):
? ? ? ? try:
? ? ? ? ? ? self.cursor.execute(sql)
? ? ? ? ? ? v_result=self.cursor.fetchall()
? ? ? ? ? ? #print(v_result)
? ? ? ? ? ? return v_result
? ? ? ? except Exception as e:
? ? ? ? ? ? print('Wrong--schemas_tables_list()')
? ? ? ? ? ? print(e)
? ? def schemas_tables_columns_list(self,sql,data):
? ? ? ? try:
? ? ? ? ? ? self.cursor.execute(sql,A=data)
? ? ? ? ? ? v_result=self.cursor.fetchall()
? ? ? ? ? ? return v_result
? ? ? ? except Exception as e:
? ? ? ? ? ? print('schemas_tables_columns_list')
? ? ? ? ? ? print(e)
? ? def schemas_tables_indexes_list(self,sql,data):
? ? ? ? try:
? ? ? ? ? ? self.cursor.execute(sql,A=data)
? ? ? ? ? ? v_result=self.cursor.fetchall()
? ? ? ? ? ? return v_result
? ? ? ? except Exception as e:
? ? ? ? ? ? print('schemas_tables_indexes_list')
? ? ? ? ? ? print(e)
? ? def close(self):
? ? ? ? self.db.close()
schemas_tables_count_sql = "SELECT COUNT(1),S.OWNER FROM DBA_TABLES S WHERE S.OWNER in ('PAY','BOSS','SETTLE','ISMP','TEMP_DSF','ACCOUNT') GROUP BY S.OWNER"
schemas_tables_list_sql ?= "SELECT S.OWNER||'.'||S.TABLE_NAME FROM DBA_TABLES S WHERE S.OWNER in ('PAY','BOSS','SETTLE','ISMP','TEMP_DSF','ACCOUNT')"
schemas_tables_columns_sql = "select a.OWNER||'.'||a.TABLE_NAME||'.'||a.COLUMN_NAME||'.'||a.DATA_TYPE||'.'||a.DATA_LENGTH from dba_tab_columns a where a.OWNER||'.'||a.TABLE_NAME = :A"
schemas_tables_indexes_sql = "SELECT T.table_owner||'.'||T.table_name||'.'||T.index_name FROM DBA_INDEXES T WHERE T.table_owner||'.'||T.table_name = :A"
jx_db ?= Oracle_Status_Output('dbadmin','QazWsx12','106.15.109.134:1522/paydb')
pro_db = Oracle_Status_Output('dbadmin','QazWsx12','localhost:1521/paydb')
jx_db.schemas_tables_count(schemas_tables_count_sql,'JX ')
pro_db.schemas_tables_count(schemas_tables_count_sql,'PRO')
jx_schemas_tables ?= jx_db.schemas_tables_list(schemas_tables_list_sql)
pro_schemas_tables = pro_db.schemas_tables_list(schemas_tables_list_sql)
#print(jx_schemas_tables)
#print(pro_schemas_tables)
def diff_jx_pro(listA,listB,listClass):
? ? if listA !=[] and listB !=[]:
? ? ? ? #listD ?= list(set(listA).union(set(listB)))
? ? ? ? listC ?= sorted(list(set(listA).intersection(set(listB))))
? ? ? ? listAC = sorted(list(set(listA).difference(set(listC))))
? ? ? ? listBC = sorted(list(set(listB).difference(set(listC))))
? ? ? ? #if sorted(listD) == sorted(listC):
? ? ? ? # ? ?print('All Tables OK')
? ? ? ? if listC == []:
? ? ? ? ? ? #print('JX ',listClass,':',listA)
? ? ? ? ? ? #print('PRO ',listClass,':',listB)
? ? ? ? ? ? print('Intersection>>','JX ',listClass,':',listAC,'--->','PRO',listClass,':',listBC)
? ? ? ? elif listAC != [] or listBC != []:
? ? ? ? ? ? print('Difference ?>>','JX ',listClass,':',listAC,'--->','PRO',listClass,':',listBC)
? ? ? ? else:
? ? ? ? ? ? pass
? ? ? ? return listC
? ? ? ?
if __name__ == '__main__':
? ? #diff_jx_pro(jx_schemas_tables,pro_schemas_tables)
? ? tables_lists = diff_jx_pro(jx_schemas_tables,pro_schemas_tables,'Tables')
? ? for i in range(len(tables_lists)):
? ? ? ? table_name = "".join(tuple(tables_lists[i]))
? ? ? ? #print(table_name)
? ? ? ? jx_schemas_tables_columns = jx_db.schemas_tables_columns_list(schemas_tables_columns_sql,table_name)
? ? ? ? pro_schemas_tables_columns = pro_db.schemas_tables_columns_list(schemas_tables_columns_sql,table_name)
? ? ? ? diff_jx_pro(jx_schemas_tables_columns,pro_schemas_tables_columns,'Columns')
? ? for i in range(len(tables_lists)):
? ? ? ? table_name = "".join(tuple(tables_lists[i]))
? ? ? ? #print(table_name)
? ? ? ? jx_schemas_tables_indexes = jx_db.schemas_tables_indexes_list(schemas_tables_indexes_sql,table_name)
? ? ? ? pro_schemas_tables_indexes = pro_db.schemas_tables_indexes_list(schemas_tables_indexes_sql,table_name)
? ? ? ? diff_jx_pro(jx_schemas_tables_indexes,pro_schemas_tables_indexes,'Indexes')
? ? jx_db.close()
? ? pro_db.close()
? ? print('------------Table, column and index check completed--------------')以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。
相關(guān)文章
Python實(shí)現(xiàn)二叉搜索樹BST的方法示例
這篇文章主要介紹了Python實(shí)現(xiàn)二叉搜索樹BST的方法示例，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-07-07
這篇文章主要介紹了python字符串分割以及字符串的一些常規(guī)方法，本文給大家介紹的非常詳細(xì)，具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2019-07-07
python+flask實(shí)現(xiàn)API的方法
這篇文章主要為大家詳細(xì)介紹了python+flask實(shí)現(xiàn)API的方法，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2018-11-11
Pytorch+PyG實(shí)現(xiàn)GraphSAGE過程示例詳解
這篇文章主要為大家介紹了Pytorch+PyG實(shí)現(xiàn)GraphSAGE過程示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2023-04-04
這篇文章主要介紹了pytorch loss反向傳播出錯(cuò)的解決方案，具有很好的參考價(jià)值，希望對大家有所幫助。一起跟隨小編過來看看吧2021-05-05
這篇文章主要介紹了Python進(jìn)程池基本概念，當(dāng)創(chuàng)建的子進(jìn)程數(shù)量不多時(shí)，可以直接利用多處理進(jìn)程中的進(jìn)程動(dòng)態(tài)形成需要的進(jìn)程，下文關(guān)于Python線程池的概念做詳細(xì)介紹，需要的小伙伴可以參考一下2022-03-03
用Python編寫一個(gè)基于終端的實(shí)現(xiàn)翻譯的腳本
這篇文章主要介紹了用Python編寫一個(gè)基于終端的實(shí)現(xiàn)翻譯的腳本,代碼基于Python2.x,需要的朋友可以參考下2015-04-04
這篇文章主要介紹了Python3+Flask安裝使用教程詳解,需要的朋友可以參考下2021-02-02
本文詳細(xì)介紹了如何使用Python實(shí)現(xiàn)兩組數(shù)據(jù)的縱向排序,包括開發(fā)思想、開發(fā)流程和代碼示例,感興趣的朋友一起看看吧2024-12-12
python opencv鼠標(biāo)交互操作的實(shí)用指南
最近的工作經(jīng)常需要對圖像進(jìn)行操作,也需要用鼠標(biāo)對圖像進(jìn)行各種操作,所以下面這篇文章主要給大家介紹了關(guān)于python opencv鼠標(biāo)交互操作的相關(guān)資料,需要的朋友可以參考下2021-10-10