postgreSQL 数据库字典导出工具

发布于:2025-06-25 ⋅ 阅读:(15) ⋅ 点赞:(0)

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


网站公告

今日签到

点亮在社区的每一天
去签到