利用Python批量导出mysql数据库表结构的操作实例_Python

前言

最近在公司售前售后同事遇到一些奇怪的需求找到我，需要提供公司一些项目数据库所有表的结构信息（字段名、类型、长度、是否主键、***、备注），虽然不是本职工作，但是作为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')