為滿足項目驗收文檔需求,開發了一個基于Python的PostgreSQL數據字典導出工具。
廢話不多說,先分享一下
軟件截圖
數據字典文件樣式,文件格式為docx
軟件源碼
基于python開發,
import tkinter as tk
from tkinter import ttk, messagebox
from PIL import Image, ImageTk
import psycopg2
from docx import Documentclass ProfessionalDBConnector:def __init__(self, root):self.root = rootself.root.title("數據字典導出工具 v1.0")self.root.geometry("500x550")self.root.resizable(False, False)self.root.configure(bg="#f5f5f5")# 加載logotry:self.logo_img = ImageTk.PhotoImage(Image.open("logo.png").resize((100, 40)))logo_label = tk.Label(root, image=self.logo_img, bg="#f5f5f5")logo_label.grid(row=0, column=0, columnspan=2, pady=(20, 30))except:title_label = tk.Label(root, text="PostgreSql數據字典導出工具", font=("微軟雅黑", 16, "bold"), bg="#f5f5f5", fg="#333")title_label.grid(row=0, column=0, columnspan=2, pady=(20, 30))# 輸入框樣式style = ttk.Style()style.configure("TLabel", background="#f5f5f5", font=("微軟雅黑", 10))style.configure("TEntry", font=("微軟雅黑", 10), padding=5)style.configure("TButton", font=("微軟雅黑", 10, "bold"), padding=5)# 輸入框標簽fields = ["主機:", "端口:", "用戶名:", "密碼:", "數據庫名:"]self.entries = {}for i, field in enumerate(fields):ttk.Label(root, text=field).grid(row=i + 1, column=0, padx=20, pady=5, sticky="e")entry = ttk.Entry(root)if field == "密碼:":entry = ttk.Entry(root, show="*")entry.grid(row=i + 1, column=1, padx=10, pady=5, sticky="ew")self.entries[field[:-1]] = entry# 連接按鈕self.connect_test_btn = tk.Button(root, text="測試連接", command=self.connect_test_db,# 設置按鈕的背景顏色為綠色bg="#4CAF50",# 設置按鈕的前景(文本)顏色為白色fg="white",# 設置按鈕在按下狀態時的背景顏色activebackground="#45a049",# 設置按鈕文本的字體和樣式font=("微軟雅黑", 10, "bold"),# 設置按鈕內部x軸方向的填充padx=15,# 設置按鈕內部y軸方向的填充pady=5,# 設置按鈕的邊框樣式為平的,無邊框relief="flat",# 設置按鈕的邊框寬度為0,與relief參數共同作用實現無邊框效果bd=0,# 設置鼠標懸停在按鈕上時的光標樣式為手型cursor="hand2")# 創建導出按鈕組件# 該按鈕用于觸發導出數據庫功能,其具體功能通過command參數關聯的self.export_db方法實現self.export_dictionary_btn = tk.Button(root, text="導出數據字典", command=self.export_dictionary,bg="#4CAF50",fg="white",activebackground="#45a049",font=("微軟雅黑", 10, "bold"),padx=15,pady=5,relief="flat",bd=0,cursor="hand2")self.connect_test_btn.grid(row=6, column=0, columnspan=2, pady=20)self.export_dictionary_btn.grid(row=7, column=0, columnspan=2, pady=20)# 底部署名和版權信息footer_frame = tk.Frame(root, bg="#e0e0e0")footer_frame.grid(row=9, column=0, columnspan=2, sticky="ew", pady=(10, 0))tk.Label(footer_frame, text="? 2025 數據庫工具 | 開發人員: guozs",bg="#e0e0e0", fg="#666", font=("微軟雅黑", 8)).pack(pady=5)# 配置網格權重root.columnconfigure(1, weight=1)def connect_test_db(self):try:conn = self.getConn()cursor = conn.cursor()cursor.execute("select version()")result = cursor.fetchall()if result is not None:messagebox.showinfo("連接成功", f"數據庫版本:{result}")conn.close()else:messagebox.showinfo("連接失敗")except Exception as e:messagebox.showerror("連接失敗", f"錯誤: {str(e)}")def getConn(self):self.database = self.entries["數據庫名"].get()if self.database == "":self.database = "abc"self.user = self.entries["用戶名"].get()if self.user == "":self.user = "postgres"self.password = self.entries["密碼"].get()if self.password == "":self.password = "123456"self.host = self.entries["主機"].get()if self.host == "":self.host = "127.0.0.1"self.port = self.entries["端口"].get()if self.port == "":self.port = "5432"conn = psycopg2.connect(database=self.database,user=self.user,password=self.password,host=self.host,port=int(self.port))return conndef export_dictionary(self):doc = Document()try:self.exportTableInfoToDocx(doc)messagebox.showinfo("成功", f"數據字典已導出")except Exception as e:messagebox.showerror("失敗", f"錯誤: {str(e)}")def getTableList(self):conn = self.getConn()cur = conn.cursor()query = '''select A.schemaname,A.relname,obj_description ( B.relfilenode, 'pg_class' ) AS tablenamefrom pg_stat_user_tables as A,pg_class as BWHERE B.relname = A.relname ORDER BYA.schemaname,A.relname'''cur.execute(query)tableList = cur.fetchall()conn.commit()cur.close()conn.close()return tableListdef getTableColumnByTableName(self, tableName):conn = self.getConn()cur = conn.cursor()tableName = "'" + tableName + "'"query = f'''SELECTd.relname AS relname,obj_description ( relfilenode, 'pg_class' ) AS tablename,attname AS field,CASEtypname WHEN '_bpchar' THEN'char' WHEN '_varchar' THEN'varchar' WHEN '_date' THEN'date' WHEN '_float8' THEN'float8' WHEN '_int4' THEN'int4' WHEN '_int8' THEN'int8' WHEN '_interval' THEN'interval' WHEN '_numeric' THEN'numeric' WHEN '_float4' THEN'float4' WHEN '_int2' THEN'smallint' WHEN '_text' THEN'text' WHEN '_time' THEN'time' WHEN '_timestamp' THEN'timestamp' WHEN '_timestamptz' THEN'timestamptz' END AS TYPE,CASEtypname WHEN '_bpchar' THENatttypmod - 4 WHEN '_varchar' THENatttypmod - 4 WHEN '_numeric' THEN( atttypmod - 4 ) / 65536 ELSE attlen END AS LENGTH,CASEtypname WHEN '_numeric' THEN( atttypmod - 4 ) % 65536 ELSE 0 END AS xs,CASEWHEN b.attnotnull = 't' THEN'不能為空' ELSE'' END AS NOTNULL,CASEWHEN ( SELECT COUNT ( * ) FROM pg_constraint WHERE conrelid = b.attrelid AND conkey [ 1 ]= attnum AND contype = 'p' ) > 0 THEN'主鍵' ELSE'' END AS zj,col_description ( b.attrelid, b.attnum ) AS COMMENT FROMpg_stat_user_tables AS A,pg_class AS d,pg_tables AS P,pg_attribute AS b,pg_type AS C WHEREA.relid = b.attrelid AND b.attnum > 0 AND b.atttypid = C.typelem AND substr( typname, 1, 1 ) = '_' AND P.tablename = d.relname AND d.relname = A.relname AND A.relname NOT LIKE'c%' AND A.relname NOT LIKE'S%' AND P.tablename = {tableName}ORDER BYA.schemaname,A.relname,attnum'''cur.execute(query)data = cur.fetchall()conn.commit()cur.close()conn.close()return datadef exportTableInfoToDocx(self, doc):tableList = self.getTableList()p = doc.add_paragraph('')table_explain = "數據字典"p.add_run(table_explain, style="Heading 1 Char")p = doc.add_paragraph('')table_explain = "數據庫名:%s" % (self.database)p.add_run(table_explain, style="Heading 1 Char")p = doc.add_paragraph('')table_explain = "表匯總"p.add_run(table_explain, style="Heading 1 Char")table = doc.add_table(rows=1, cols=3)table.style = 'TableGrid'hdr_cells = table.rows[0].cellshdr_cells[0].text = '模式'hdr_cells[1].text = '表名'hdr_cells[2].text = '表注釋'for tableInfo in tableList:new_cells = table.add_row().cellsnew_cells[0].text = tableInfo[0]new_cells[1].text = tableInfo[1]# 判斷注釋是否為空new_cells[2].text = getStr(tableInfo[2]) # if tableInfo[2] is None else tableInfo[2]p = doc.add_paragraph('')p = doc.add_paragraph('')p = doc.add_paragraph('')table_explain = "表詳情"p.add_run(table_explain, style="Heading 1 Char")try:for tableInfo in tableList:tableName = tableInfo[1]tableComment = getStr(tableInfo[2])p = doc.add_paragraph('')table_explain = "表名:%s, 注解:%s" % (tableName, getStr(tableComment)) # tableName + ",注解:" + getInfo(tableComment) + ",對應數據庫的表:"p = doc.add_paragraph('')p.add_run(table_explain, style="Heading 1 Char")table = doc.add_table(rows=1, cols=5)table.style = 'TableGrid'hdr_cells = table.rows[0].cellshdr_cells[0].text = '字段名'hdr_cells[1].text = '字段類型'hdr_cells[2].text = '允許為空'hdr_cells[3].text = 'PK'hdr_cells[4].text = '字段說明'tableColumnList = self.getTableColumnByTableName(tableInfo[1])for tableColumn in tableColumnList:new_cells = table.add_row().cellsnew_cells[0].text = getStr(tableColumn[2])new_cells[1].text = getStr(tableColumn[3])new_cells[2].text = getStr(tableColumn[6])new_cells[3].text = getStr(tableColumn[7])new_cells[4].text = getStr(tableColumn[8])except Exception as e:messagebox.showerror("失敗", f"錯誤: {str(e)}")p = doc.add_paragraph('')doc.save('./' + self.database + '_數據字典.docx')def getStr(param):if param is None:return ""else:return paramif __name__ == "__main__":root = tk.Tk()app = ProfessionalDBConnector(root)root.mainloop()