为满足项目验收文档需求,开发了一个基于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()