利用Python批量導出mysql數據庫表結構的操作實例

Posted on 2022-08-08 by WalkonNet

前言

最近在公司售前售後同事遇到一些奇怪的需求找到我，需要提供公司一些項目數據庫所有表的結構信息（字段名、類型、長度、是否主鍵、***、備註），雖然不是本職工作，但是作為python技能的擁有者看到這種需求還是覺得很容易的，但是如果不用代碼解決確實非常棘手和浪費時間。於是寫瞭一個輕量小型項目來解決一些燃眉之急，希望能對一些人有所幫助，代碼大神、小神可以忽略此貼。

代碼直達: GITEE、GitHub

解決方法

1. mysql 數據庫表信息查詢

想要導出mysql數據庫表結構必須瞭解一些相關數據庫知識，mysql數據庫支持通過SQL語句進行表信息查詢：

查詢數據庫所有表名

SHOW TABLES

查詢對應數據庫對應表結構信息

SELECT COLUMN_NAME,COLUMN_TYPE,COLUMN_KEY,IS_NULLABLE, COLUMN_COMMENT 
FROM information_schema.`COLUMNS` 
WHERE TABLE_SCHEMA='{dbName}' AND TABLE_NAME='{tableName}'

COLUMN_NAME：字段名
COLUMN_TYPE：數據類型
COLUMN_KEY：主鍵
IS_NULLABLE：非空
COLUMN_COMMENT：字段描述
還有一些其他字段，有需要可自行百度

2.連接數據庫代碼

以下是一個較為通用的mysql數據庫連接類，創建 MysqlConnection 類，放入對應數據庫連接信息即可使用sql,通過query查詢、update增刪改、close關閉連接。

*註：數據量過大時不推薦直接使用query查詢。

import pymysql

class MysqlConnection():
    def __init__(self, host, user, passw, port, database, charset="utf8"):
        self.db = pymysql.connect(host=host, user=user, password=passw, port=port,
                                  database=database, charset=charset)
        self.cursor = self.db.cursor()

    # 查
    def query(self, sql):
        self.cursor.execute(sql)
        results = self.cursor.fetchall()
        return results

    # 增刪改
    def update(self, sql):
        try:
            self.cursor.execute(sql)
            self.db.commit()
            return 1
        except Exception as e:
            print(e)
            self.db.rollback()
            return 0

    # 關閉連接
    def close(self):
        self.cursor.close()
        self.db.close()

3.數據查詢處理代碼

3.0 配置信息

config.yml,這裡使用瞭配置文件進行程序參數配置，方便配置一鍵運行

# 數據庫信息配置
db_config:
  host: 127.0.0.1	# 數據庫所在服務IP
  port: 3306		# 數據庫服務端口
  username: root	# ~用戶名
  password: 12346	# ~密碼
  charset: utf8
  # 需要進行處理的數據名稱列表 《《 填入數據庫名
  db_names: ['db_a','db_b']

# 導出配置
excel_conf:
  # 導出結構Excel表頭，長度及順序不可調整，僅支持更換名稱
  column_name: ['字段名', '數據類型', '長度', '主鍵', '非空', '描述']
  save_dir: ./data

讀取配置文件的代碼

import yaml

class Configure():
    def __init__(self):
        with open("config.yaml", 'r', encoding='utf-8') as f:
            self._conf = yaml.load(f.read(), Loader=yaml.FullLoader)

    def get_db_config(self):
        host = self._conf['db_config']['host']
        port = self._conf['db_config']['port']
        username = self._conf['db_config']['username']
        password = self._conf['db_config']['password']
        charset = self._conf['db_config']['charset']
        db_names = self._conf['db_config']['db_names']
        return host, port, username, password, charset, db_names

    def get_excel_title(self):
        title = self._conf['excel_conf']['column_name']
        save_dir = self._conf['excel_conf']['save_dir']
        return title, save_dir

3.1查詢數據庫表

利用上面創建的數據庫連接和SQL查詢獲取所有表

class ExportMysqlTableStructureInfoToExcel():
	def __init__(self):
	        conf = Configure()	# 獲取配置初始化類信息
	        self.__host, self.__port, self.__username, self.__password, self.__charset, self.db_names = conf.get_db_config()
	        self.__excel_title, self.__save_dir = conf.get_excel_title()
	```省略```
	def __connect_to_mysql(self, database):	# 獲取數據庫連接方法
        connect = MysqlConnection(self.__host,
                                  self.__username,
                                  self.__password,
                                  self.__port, database,
                                  self.__charset)
        return connect
        
	def __get_all_tables(self, con):	# 查詢所有表
	        res = con.query("SHOW TABLES")
	        tb_list = []
	        for item in res:
	            tb_list.append(item[0])
	        return tb_list
	``````

3.2 查詢對應表結構

循環獲取每一張表的結構數據，根據需要對中英文做瞭一些轉換，字段長度可以從類型中分離出來，這裡使用yield返回數據，可以利用生成器加速處理過程（外包導出保存和數據庫查詢可以並行）

class ExportMysqlTableStructureInfoToExcel():
	```省略```
	def __struct_of_table_generator(self, con, db_name):
        tb_list = self.__get_all_tables(con)
        for index, tb_name in enumerate(tb_list):
            sql = "SELECT COLUMN_NAME,COLUMN_TYPE,COLUMN_KEY,IS_NULLABLE, COLUMN_COMMENT " \
              "FROM information_schema.`COLUMNS` WHERE TABLE_SCHEMA='{}' AND TABLE_NAME='{}'".format(db_name, tb_name)
            res = con.query(sql)
            struct_list = []
            for item in res:
                column_name, column_type, column_key, is_nullable, column_comment = item
                length = "0"
                if str(column_type).find('(') > -1:
                    column_type, length = str(column_type).replace(")", '').split('(')
                if column_key == 'PRI':
                    column_key = "是"
                else:
                    column_key = ''
                if is_nullable == 'YES':
                    is_nullable = '是'
                else:
                    is_nullable = '否'
                struct_list.append([column_name, column_type, length, column_key, is_nullable, column_comment])
            yield [struct_list, tb_name]
	```省略```

3.3 pandas進行數據保存導出excel

class ExportMysqlTableStructureInfoToExcel():
	```省略```
	def export(self):
        if len(self.db_names) == 0:
            print("請配置數據庫列表")
        for i, db_name in enumerate(self.db_names):		# 對多個數據庫進行處理
            connect = self.__connect_to_mysql(db_name)	# 獲取數據庫連接
            if not os.path.exists(self.__save_dir):		# 判斷數據導出保存路徑是否存在
                os.mkdir(self.__save_dir)

            file_name = os.path.join(self.__save_dir,'{}.xlsx'.format(db_name))	# 用數據庫名命名導出Excel文件
            if not os.path.exists(file_name):  # 文件不存在時自動創建文件 excel
                wrokb = openpyxl.Workbook()
                wrokb.save(file_name)
                wrokb.close()
            wb = openpyxl.load_workbook(file_name)
            writer = pd.ExcelWriter(file_name, engine='openpyxl')
            writer.book = wb

            struct_generator = self.__struct_of_table_generator(connect, db_name)	# 獲取表結構信息的生成器

            for tb_info in tqdm(struct_generator, desc=db_name):	# 從生成器中獲取表結構並利用pandas進行格式化保存，寫入Excel文件
                s_list, tb_name = tb_info
                data = pd.DataFrame(s_list, columns=self.__excel_title)
                data.to_excel(writer, sheet_name=tb_name)
            writer.close()

            connect.close()
	```省略```

補充：python腳本快速生成mysql數據庫結構文檔

由於數據表太多，手動編寫耗費的時間太久，所以搞瞭一個簡單的腳本快速生成數據庫結構，保存到word文檔中。

1.安裝pymysql和document

pip install pymysql
pip install document

2.腳本

# -*- coding: utf-8 -*-
import pymysql
from docx import Document
from docx.shared import Pt
from docx.oxml.ns import qn

db = pymysql.connect(host='127.0.0.1', #數據庫服務器IP
                         port=3306,
                         user='root',
                         passwd='123456',
                         db='test_db') #數據庫名稱)
#根據表名查詢對應的字段相關信息
def query(tableName):
    #打開數據庫連接
    cur = db.cursor()
    sql = "select b.COLUMN_NAME,b.COLUMN_TYPE,b.COLUMN_COMMENT from (select * from information_schema.`TABLES`  where TABLE_SCHEMA='test_db') a right join(select * from information_schema.`COLUMNS` where TABLE_SCHEMA='test_db_test') b on a.TABLE_NAME = b.TABLE_NAME where a.TABLE_NAME='" + tableName+"'"
    cur.execute(sql)
    data = cur.fetchall()
    cur.close
    return data
#查詢當前庫下面所有的表名，表名：tableName；表名+註釋(用於填充至word文檔)：concat(TABLE_NAME,'(',TABLE_COMMENT,')')
def queryTableName():
    cur = db.cursor()
    sql = "select TABLE_NAME,concat(TABLE_NAME,'(',TABLE_COMMENT,')') from information_schema.`TABLES`  where TABLE_SCHEMA='test_db_test'"
    cur.execute(sql)
    data = cur.fetchall()
    return data
#將每個表生成word結構，輸出到word文檔
def generateWord(singleTableData,document,tableName):
    p=document.add_paragraph()
    p.paragraph_format.line_spacing=1.5 #設置該段落 行間距為 1.5倍
    p.paragraph_format.space_after=Pt(0) #設置段落 段後 0 磅
    #document.add_paragraph(tableName,style='ListBullet')
    r=p.add_run('\n'+tableName)
    r.font.name=u'宋體'
    r.font.size=Pt(12)
    table = document.add_table(rows=len(singleTableData)+1, cols=3,style='Table Grid')
    table.style.font.size=Pt(11)
    table.style.font.name=u'Calibri'
    #設置表頭樣式
    #這裡隻生成瞭三個表頭，可通過實際需求進行修改
    for i in ((0,'NAME'),(1,'TYPE'),(2,'COMMENT')):
        run = table.cell(0,i[0]).paragraphs[0].add_run(i[1])
        run.font.name = 'Calibri'
        run.font.size = Pt(11)
        r = run._element
        r.rPr.rFonts.set(qn('w:eastAsia'), '宋體')
    
    for i in range(len(singleTableData)):
        #設置表格內數據的樣式
        for j in range(len(singleTableData[i])):
            run = table.cell(i+1,j).paragraphs[0].add_run(singleTableData[i][j])
            run.font.name = 'Calibri'
            run.font.size = Pt(11)
            r = run._element
            r.rPr.rFonts.set(qn('w:eastAsia'), '宋體')
        #table.cell(i+1, 0).text=singleTableData[i][1]
        #table.cell(i+1, 1).text=singleTableData[i][2]
        #table.cell(i+1, 2).text=singleTableData[i][3]
    

if __name__ == '__main__':
    #定義一個document
    document = Document()
    #設置字體默認樣式
    document.styles['Normal'].font.name = u'宋體'
    document.styles['Normal']._element.rPr.rFonts.set(qn('w:eastAsia'), u'宋體')
    #獲取當前庫下所有的表名信息和表註釋信息
    tableList = queryTableName()
    #循環查詢數據庫，獲取表字段詳細信息，並調用generateWord，生成word數據
    #由於時間匆忙，我這邊選擇的是直接查詢數據庫，執行瞭100多次查詢，可以進行優化，查詢出所有的表結構，在代碼裡面將每個表結構進行拆分
    for singleTableName in tableList:
        data = query(singleTableName[0])
        generateWord(data,document,singleTableName[1])
    #保存至文檔
    document.save('數據庫設計.docx')

3.生成的word文檔預覽

總結

運行成功後會在目錄下的data文件夾中看到保存的Excel文件（以數據庫名為單位保存成文件），每個Excel第一個tab是空的（一個小bug暫未解決），其他每個tab以對應表名進行命名。

代碼很簡單，供各位學習參考。

到此這篇關於利用Python批量導出mysql數據庫表結構的文章就介紹到這瞭,更多相關Python批量導出mysql表結構內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

利用Python批量導出mysql數據庫表結構的操作實例

目錄

前言

解決方法

1. mysql 數據庫表信息查詢

2.連接數據庫代碼

3.數據查詢處理代碼

3.0 配置信息

3.1查詢數據庫表

3.2 查詢對應表結構

3.3 pandas進行數據保存導出excel

補充：python腳本快速生成mysql數據庫結構文檔

總結

推薦閱讀：

發佈留言取消回覆

近期文章

目錄

前言

解決方法

1. mysql 數據庫 表信息查詢

2.連接數據庫代碼

3.數據查詢處理代碼

3.0 配置信息

3.1查詢數據庫表

3.2 查詢對應表結構

3.3 pandas進行數據保存導出excel

補充：python腳本快速生成mysql數據庫結構文檔

總結

推薦閱讀：

發佈留言 取消回覆

近期文章

標籤

1. mysql 數據庫表信息查詢

發佈留言取消回覆