为满足项目验收文档需求,开发了一个基于Python的PostgreSQL数据字典导出工具。
废话不多说,先分享一下
软件截图
数据字典文件样式,文件格式为docx
软件源码
基于python开发,
import tkinter as tk
from tkinter import ttk, messagebox
from PIL import Image, ImageTk
import psycopg2
from docx import Document
class ProfessionalDBConnector:
def __init__(self, root):
self.root = root
self.root.title("数据字典导出工具 v1.0")
self.root.geometry("500x550")
self.root.resizable(False, False)
self.root.configure(bg="#f5f5f5")
# 加载logo
try:
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()
info = conn.info
if info is not None:
messagebox.showinfo("连接成功", f"数据库连接成功!")
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"
return psycopg2.connect(database=self.database,
user=self.user,
password=self.password,
host=self.host,
port=int(self.port)
)
def 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 tablename
from
pg_stat_user_tables as A,
pg_class as B
WHERE
B.relname = A.relname
ORDER BY
A.schemaname,
A.relname
'''
cur.execute(query)
tableList = cur.fetchall()
conn.commit()
cur.close()
conn.close()
return tableList
def getTableColumnByTableName(self, tableName):
conn = self.getConn()
cur = conn.cursor()
tableName = "'" + tableName + "'"
query = f'''
SELECT
d.relname AS relname,
obj_description ( relfilenode, 'pg_class' ) AS tablename,
attname AS field,
CASE
typname
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,
CASE
typname
WHEN '_bpchar' THEN
atttypmod - 4
WHEN '_varchar' THEN
atttypmod - 4
WHEN '_numeric' THEN
( atttypmod - 4 ) / 65536 ELSE attlen
END AS LENGTH,
CASE
typname
WHEN '_numeric' THEN
( atttypmod - 4 ) % 65536 ELSE 0
END AS xs,
CASE
WHEN b.attnotnull = 't' THEN
'不能为空' ELSE''
END AS NOTNULL,
CASE
WHEN ( 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
FROM
pg_stat_user_tables AS A,
pg_class AS d,
pg_tables AS P,
pg_attribute AS b,
pg_type AS C
WHERE
A.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 BY
A.schemaname,
A.relname,
attnum
'''
cur.execute(query)
data = cur.fetchall()
conn.commit()
cur.close()
conn.close()
return data
def 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].cells
hdr_cells[0].text = '模式'
hdr_cells[1].text = '表名'
hdr_cells[2].text = '表注释'
for tableInfo in tableList:
new_cells = table.add_row().cells
new_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")
for tableInfo in tableList:
tableName = tableInfo[1]
tableComment = 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].cells
hdr_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().cells
new_cells[0].text = tableColumn[2]
new_cells[1].text = tableColumn[3]
new_cells[2].text = tableColumn[6]
new_cells[3].text = getStr(tableColumn[7])
new_cells[4].text = tableColumn[8]
p = doc.add_paragraph('')
doc.save('./' + self.database + '_数据字典.docx')
def getStr(param):
if param is None:
return "无"
else:
return param
if __name__ == "__main__":
root = tk.Tk()
app = ProfessionalDBConnector(root)
root.mainloop()