[数据库管理]SQL表定义查询与数据字典的导出

最后一次更新日期：2019-5-27

分享一段简短的 python 脚本，可用于查询 sql 数据库表定义以及导出数据字典excel文档。

支持 mysql 和mssql(SQL Server)，由于本人使用mysql较少，所以mysql的查询内容相对简单，如有需要可以自行增加。

效果图：

完整代码：

# -*- coding: utf-8 -*-
import pandas as pd
import sqlalchemy as sqla
from openpyxl import load_workbook
from openpyxl.styles import Font,colors,Border,Side,Alignment,PatternFill
   
class SqlDataDictionary:
    '''\n
    Class: SQL数据字典
    
    Parameters
    ----------
    dbtype: 数据库类型，仅支持'mysql'和'mssql'，
            需要安装sqlalchemy + pymysql或pymssql
    username: 用户名
    password: 密码
    server: 服务器地址
    database: 数据库名，默认None，会连接默认数据库
    charset: 字符集，默认'utf8'
    engine: sqlalchemy的连接引擎，设置此项可以代替前面所有的参数
    ----------
    
    Attributes
    ----------
    engine_: sqlalchemy的连接引擎
    excel_col_width_: excel文档的列宽设置
    excel_border_: excel文档的边框设置
    excel_font_: excel文档的字体设置
    excel_align_: excel文档的对齐设置
    excel_fill_: excel文档的填充设置
    ----------
    '''
    def __init__(self,dbtype=None,username=None,password=None,server=None,
                 database=None,charset='utf8',engine=None):
        
        #生成engine
        if type(engine)==type(None):
            engine=self._create_engine(dbtype,username,password,server,database,charset)
        self.engine_=engine
        
        #excel文档列设置
        self.excel_col_width_={
                'TABLE_CATALOG': 20,
                'TABLE_SCHEMA': 20,
                'TABLE_NAME': 30,
                'TABLE_TYPE': 20,
                'TABLE_COMMENT': 40,
                'TABLE_ROWS': 20,
                'CREATE_TIME': 20,
                'UPDATE_TIME': 20,
                'ORDINAL': 10,
                'COLUMN_NAME': 25,
                'COLUMN_COMMENT': 35,
                'COLUMN_TYPE': 15,
                'COLUMN_DEFAULT': 20,
                'COLUMN_KEY': 15,
                'IS_NULLABLE': 15,
                'IS_IDENTITY': 15,
                'IS_UNIQUEKEY': 15,
                'IS_PRIMARYKEY': 15,
                'IS_COMPUTED': 15,
                'EXPRESSION': 20,
                'INDEX_TYPE': 15,
                'KEY_ORDINAL': 15
                }
        
        #excel文档边框线设置
        self.excel_border_={
                'all': Border(
                        left=Side(style='medium',color='FF000000'),
                        right=Side(style='medium',color='FF000000'),
                        top=Side(style='medium',color='FF000000'),
                        bottom=Side(style='medium',color='FF000000'),
                        diagonal=Side(style='medium',color='FF000000'),
                        diagonal_direction=0,
                        outline=Side(style='medium',color='FF000000'),
                        vertical=Side(style='medium',color='FF000000'),
                        horizontal=Side(style='medium',color='FF000000')
                        )
                }
                
        #excel文档字体设置
        self.excel_font_={
                'link': Font(underline='single',color=colors.BLUE),
                'head': Font(bold=True)
                }
        #excel文档对齐设置
        self.excel_align_={
                'center':Alignment(horizontal='center')
                }
        #excel文档填充设置
        self.excel_fill_={
                'link':PatternFill(fill_type='solid',start_color='E6E6E6'),
                'head':PatternFill(fill_type='solid',start_color='CDDCE6')
                }
    
    #创建sqlalchemy连接引擎
    def _create_engine(self,dbtype,username,password,server,database,charset):
        #设置默认数据库
        if type(database)==type(None):
            if dbtype=='mysql':
                database='sys'
            elif dbtype=='mssql':
                database='master'
            #此处可拓展其他数据库
            else:
                raise Exception('unsupported dbtype')
                
        engine=sqla.create_engine(
                '{}+py{}://{}:{}@{}/{}?charset={}'
                .format(dbtype,dbtype,username,password,server,database,charset)
                )
        return engine
    
    #变更数据库
    def change_database(self,database):
        
        self.engine_.url.database=database
        self.engine_.dispose()
        self.engine_=sqla.create_engine(self.engine_.url)
        print('database is changed to '+database)
    
    #查询数据库名列表(mysql)
    def _query_schema_databases_mysql(self):
        
        databases=pd.read_sql('show databases;',self.engine_)
        databases=databases['Database'].tolist()
        
        return databases
    
    #查询表定义(mysql)
    def _query_schema_tables_mysql(self):
        
        tables=pd.read_sql('''
        select TABLE_NAME,TABLE_COMMENT,TABLE_TYPE,
            CREATE_TIME,TABLE_ROWS 
        from information_schema.TABLES 
        where TABLE_SCHEMA=database()
        order by 1;
        ''',self.engine_)
        
        columns=pd.read_sql('''
        select TABLE_NAME,ORDINAL_POSITION ORDINAL,COLUMN_NAME,
            COLUMN_COMMENT, COLUMN_TYPE,COLUMN_DEFAULT,COLUMN_KEY,IS_NULLABLE 
        from information_schema.COLUMNS 
        where TABLE_SCHEMA=database()
        order by 1,2;
        ''',self.engine_)
        
        return tables,columns
    
    #查询数据库名列表(sql server)
    def _query_schema_databases_mssql(self):
        
        databases=pd.read_sql('select name from sys.databases',self.engine_)
        databases=databases['name'].tolist()
        
        return databases
    
    #查询表定义(sql server)
    def _query_schema_tables_mssql(self):

        tables=pd.read_sql('''
        SELECT ss.name+'.'+so.name TABLE_NAME,CONVERT(NVARCHAR(4000),sep.value) TABLE_COMMENT,
            so.type_desc TABLE_TYPE,so.create_date CREATE_TIME,si2.rows TABLE_ROWS
        FROM sys.objects so
        JOIN sys.schemas ss ON so.schema_id=ss.schema_id
        JOIN sysindexes si2 ON so.object_id=si2.id AND si2.indid<2
        LEFT JOIN sys.extended_properties sep ON so.object_id=sep.major_id AND sep.minor_id=0 AND sep.name='MS_Description' 
        WHERE so.type IN ('U','V')
        ORDER BY 1
        ''',self.engine_)
        
        columns=pd.read_sql('''
        SELECT ss.name+'.'+so.name TABLE_NAME,sc.column_id ORDINAL,
            sc.name COLUMN_NAME,CONVERT(NVARCHAR(4000),sep.value) COLUMN_COMMENT,
            case when sc.max_length=-1 then st.name+'(max)'
                when st.name in ('nchar','nvarchar') then st.name+'('+CAST(sc.max_length/2 as varchar(10))+')'
                when st.name in ('char','varchar','binary','varbinary') then st.name+'('+CAST(sc.max_length as varchar(10))+')'
                when st.name in ('numeric','decimal') then st.name+'('+CAST(sc.precision as varchar(5))+','+CAST(sc.scale as varchar(5))+')'
                else st.name end COLUMN_TYPE,
            sdc.definition COLUMN_DEFAULT,sc.is_nullable IS_NULLABLE,sc.is_identity IS_IDENTITY,
            ISNULL(si.is_unique_constraint,0) IS_UNIQUEKEY,ISNULL(si.is_primary_key,0) IS_PRIMARYKEY,
            sc.is_computed IS_COMPUTED,scc.definition EXPRESSION,
            si.type_desc INDEX_TYPE,sic.key_ordinal KEY_ORDINAL
        FROM sys.objects so
        JOIN sys.schemas ss ON so.schema_id=ss.schema_id
        JOIN sys.columns sc ON so.object_id=sc.object_id
        JOIN sys.types st ON sc.user_type_id=st.user_type_id
        LEFT JOIN sys.default_constraints sdc ON sc.default_object_id=sdc.object_id
        LEFT JOIN sys.extended_properties sep ON so.object_id=sep.major_id AND sc.column_id=sep.minor_id AND sep.name='MS_Description'
        LEFT JOIN sys.index_columns sic ON sic.object_id=so.object_id AND sic.column_id=sc.column_id
        LEFT JOIN sys.indexes si ON sic.object_id=si.object_id AND sic.index_id=si.index_id
        LEFT JOIN sys.computed_columns scc ON sc.object_id=scc.object_id AND sc.column_id=scc.column_id
        WHERE so.type in ('U','V')
        ORDER BY 1,2
        ''',self.engine_)
        
        return tables,columns
    
    #查询数据库名列表
    def query_schema_databases(self):
        '''\n
        Method: 查询数据库名列表
        '''
        if self.engine_.name=='mysql':
            databases=self._query_schema_databases_mysql()
        elif self.engine_.name=='mssql':
            databases=self._query_schema_databases_mssql()
        #此处可拓展其他数据库
        else:
            raise Exception('unsupported dbtype')
            
        return databases
    
    #查询表定义
    def query_schema_tables(self,database=None):
        '''\n
        Method: 查询表定义
        
        Parameters
        ----------
        database: 需要查询所有表定义的数据库，str类型，
                  None表示使用engine中的设置
        ----------
        
        Returns
        ----------
        tables: 表信息，DataFrame类型
        columns: 列信息，DataFrame类型
        ----------
        '''
        if type(database)!=type(None):
            self.change_database(database)
        
        if self.engine_.name=='mysql':
            tables,columns=self._query_schema_tables_mysql()
        elif self.engine_.name=='mssql':
            tables,columns=self._query_schema_tables_mssql()
        #此处可拓展其他数据库
        else:
            raise Exception('unsupported dbtype')
            
        if (not tables.columns.contains('TABLE_NAME')) or \
            (not columns.columns.contains('TABLE_NAME')):
            raise Exception("missing column 'TABLE_NAME'")
            
        return tables,columns

    #Excel列索引转标签
    def _excel_col_label(self,idx):
        
        label=''
        while True:
            label+=chr(idx%26+65)
            idx=idx//26
            if idx==0:
                break;
                
        return label[::-1]
    
    #导出Excel文档
    def to_excel(self,output_folder,databases=None,name_prefix=''):
        '''\n
        Method: 导出Excel文档
        
        Parameters
        ----------
        output_folder: 输出文件夹的路径，str类型
        databases: 需要查询所有表定义的数据库，
                   None表示使用engine中的设置，
                   str类型指定一系列逗号分隔的数据库名或'*'表示导出所有，
                   list of str类型指定数据库名列表
        name_prefix: 导出文件名的前缀（数据库名会作为后缀）
        ----------
        '''
        if name_prefix!='': name_prefix+='_'
        
        if type(databases)==type(None):
            databases=self.engine_.url.database
        
        if type(databases)==str:
            if databases=='*':
                databases=self.query_schema_databases()
            else:
                databases=[db for db in databases.split(',') if db!='']
        
        if type(databases)==list:
            for database in databases:
                tables,columns=self.query_schema_tables(database)
                file_path=output_folder+'\\'+name_prefix+database+'.xlsx'
                self.schema_to_excel(tables,columns,file_path)
        else:
            raise Exception('databases should be None(use engine setting),'+ 
                            'str(database names or ''*''), '+
                            'or list of str(database names)')
    
    #将架构信息导出至单个Excel文档
    def schema_to_excel(self,tables,columns,file_path):
        '''\n
        Method: 将架构信息导出至单个Excel文档
        
        Parameters
        ----------
        tables: 表信息，DataFrame类型
        columns: 列信息，DataFrame类型
        file_path: excel文件路径
        ----------
        '''
        columns_0=columns['TABLE_NAME']
        columns_1=columns.drop('TABLE_NAME',axis=1)
    
        #导出数据至Excel
        writer=pd.ExcelWriter(file_path)
        tables.to_excel(writer,'Index',index=False)
        for i in range(tables.shape[0]):
            table_name=tables['TABLE_NAME'].iloc[i]
            columns_=columns_1[columns_0==table_name]
            columns_.to_excel(writer,'Table'+str(i+1),index=False)
        writer.save()
        
        #调整索引页格式
        wb=load_workbook(file_path) 
        ws=wb["Index"]
        
        #调整列宽
        for j in range(tables.shape[1]):
            label=self._excel_col_label(j)
            width=self.excel_col_width_[tables.columns[j]]
            ws.column_dimensions[label].width=width
            ws[label+'1'].fill=self.excel_fill_['head']
            
        #增加边框线
        for i in range(1,tables.shape[0]+2):
            for j in range(tables.shape[1]):
                label=self._excel_col_label(j)
                ws[label+str(i)].border=self.excel_border_['all']
                
        #处理各表的列定义页
        for i in range(len(tables)):
            
            table_name=tables['TABLE_NAME'].iloc[i]
            sheet_name='Table'+str(i+1)
            
            #索引页增加调转指定表页的链接
            jump_link_colidx=tables.columns.tolist().index('TABLE_NAME')
            jump_link_cell=self._excel_col_label(jump_link_colidx)+str(i+2)
            back_link_colidx=columns.shape[1]-1
            back_link_collab=self._excel_col_label(back_link_colidx)
            back_link_cell=back_link_collab+'3'
            ws[jump_link_cell].hyperlink = "#"+sheet_name+"!A1"
            ws[jump_link_cell].font = self.excel_font_['link']

            #指定表页增加返回索引页的链接
            ws2=wb[sheet_name]
            ws2[back_link_cell]='back'
            ws2[back_link_cell].hyperlink = "#Index!"+jump_link_cell
            ws2[back_link_cell].font=self.excel_font_['link']
            ws2[back_link_cell].border=self.excel_border_['all']
            ws2[back_link_cell].alignment =self.excel_align_['center']
            ws2[back_link_cell].fill=self.excel_fill_['link']
            
            ws2.column_dimensions[back_link_collab].width=40
            
            #添加表名信息
            tname_head_cell=back_link_collab+'1'
            tname_value_cell=back_link_collab+'2'
            ws2[tname_head_cell]='TABLE_NAME'
            ws2[tname_head_cell].font =self.excel_font_['head']
            ws2[tname_head_cell].alignment =self.excel_align_['center']
            ws2[tname_head_cell].border=self.excel_border_['all']
            ws2[tname_head_cell].fill=self.excel_fill_['link']
            ws2[tname_value_cell]=table_name
            ws2[tname_value_cell].alignment =self.excel_align_['center']
            ws2[tname_value_cell].border=self.excel_border_['all']
            ws2[tname_value_cell].fill=self.excel_fill_['link']
            
            #筛选指定表的列定义
            columns_=columns_1[columns_0==table_name]
            
            #调整列宽
            for j in range(columns_.shape[1]):
                label=self._excel_col_label(j)
                width=self.excel_col_width_[columns_.columns[j]]
                ws2.column_dimensions[label].width=width
                ws2[label+'1'].fill=self.excel_fill_['head']
            
            #增加边框线
            for i in range(1,columns_.shape[0]+2):
                for j in range(columns_.shape[1]):
                    label=self._excel_col_label(j)
                    ws2[label+str(i)].border=self.excel_border_['all']
        
        #保存文件
        wb.save(file_path)
        wb.close()
        
    #读取Excel文档
    def read_excel(self,file_path):
        '''\n
        Method: 读取Excel文档
        
        Parameters
        ----------
        file_path: excel文件路径
        ----------
        
        Returns
        ----------
        tables: 表信息，DataFrame类型
        columns: 列信息，DataFrame类型
        ----------
        '''
        data=pd.read_excel(file_path, None)
        
        columns=[]
        for sheetname in data.keys():
            if sheetname=='Index':
                tables=data[sheetname]
            else:
                columns.append(data[sheetname])
        
        columns=pd.concat(columns)
        columns.insert(0,'TABLE_NAME',columns.pop('TABLE_NAME'))
        columns.index=range(columns.shape[0])
        columns.loc[columns['TABLE_NAME']=='back','TABLE_NAME']=None
        columns['TABLE_NAME']=columns['TABLE_NAME'].fillna(method='ffill')
                
        return tables,columns
  
#示例      
if __name__=='__main__':
    
    #mysql
    dbtype='mysql'
    username='root'
    password='123456'
    server='localhost:3306'
    '''
    #mssql
    dbtype='mssql'
    username='sa'
    password='123456'
    server='localhost:1433'
    
    #mssql使用windows账户连接
    engine=sqla.create_engine("mssql+pyodbc://localhost/master"+
                              "?driver=SQL Server Native Client 11.0")
    data_dict=SqlDataDictionary(engine=engine)
    '''
    #print(data_dict.query_schema_databases())
    #print(data_dict.query_schema_tables())
    
    data_dict=SqlDataDictionary(dbtype,username,password,server)
    
    data_dict.to_excel(output_folder='C:\\Users\\hp\\Desktop', 
                       databases='world', name_prefix='local')
    
    #tables,columns=data_dict.read_excel('C:\\Users\\hp\\Desktop\\local_world.xlsx')
复制代码

说明：

该小 工具 主要有三部分：

1. 从数据库系统视图中查询表和字段定义；
2. 通过 sqlalchemy 和 pandas 执行查询 sql 并将结果以 DataFrame 返回；
3. 通过 pandas 导出数据至 Excel 文件，再使用 openpyxl 调整 Excel 的格式。

查询表和字段定义

mysql 的查询使用了系统数据库 information_schema 下的系统视图，该数据库包含所有数据库的定义信息，查询时注意限制当前连接数据库。

show databases;

use db_name
select * from information_schema.TABLES where TABLE_SCHEMA=database();
select * from information_schema.COLUMNS where TABLE_SCHEMA=database();
复制代码

mssql 的查询使用了 sys 架构下的系统视图，包含当前连接数据库的定义信息。 mssql 也有 information_schema 架构的系统视图，大部分字段和 mysql 中的是一样的，但此处没有使用。

--数据库
select * from sys.databases
--数据库对象(表、视图、函数、存储过程、约束等)
--通过限制type属性查询表(U)和视图(V)
select * from sys.objects where type in ('U','V')
--列
select * from sys.columns
--类型
select * from sys.types
--所有者架构
select * from sys.schemas
--拓展属性
--通过限制name='MS_Description'查询表和列的描述
--major_id对应object_id，minor_id对应column_id，
--minor_id=0表示是表的描述
select * from sys.extended_properties
--索引
select * from sys.indexes
--索引列
select * from sys.index_columns
--计算列
select * from sys.computed_columns
--默认值约束
select * from sys.default_constraints
复制代码

有一点要注意， mssql 和 mysql 的组织结构不太一样： mssql 的数据库对应的是 CATELOG ，所有者对应着 SCHEMA ；而 mysql 的数据库对应的是 SCHEMA ， CATELOG 全为 def 。

使用 sqlalchemy + pandas 执行 sql 查询

pandas 的 read_sql 方法用于从 sql 查询读取数据生成 DataFrame ：

第一个参数 sql 设置要执行的 sql ；

第二个参数 con 设置数据库连接，主要支持类型为 sqlalchemy 的 engine 或 connection 。

import pandas as pd
result=pd.read_sql('select * from test',engine)
复制代码

sqlalchemy 的 create_engine 方法用于创建 engine 对象，连接 url 遵循 RFC-1738 ，常见形式如下：

import sqlalchemy as sqla
engine=sqla.create_engine('{dbtype}+{driver}://{username}:{password}@{server}/{database}'+
                          '?charset={charset}'')
复制代码

使用 windows authentication 连接 mssql 需要使用支持该认证方式的驱动，比如 pyodbc ，示例如下：

engine=sqla.create_engine("mssql+pyodbc://localhost/master"+
                          "?driver=SQL Server Native Client 11.0")
复制代码

创建基于 pymysql 的 engine 后首次执行 sql 查询可能会出现如下警告，但查询结果正常。

经查找资料确定这是 mysql 的一个 bug ，据说不影响使用（尚不能肯定）。

C:\ProgramData\Anaconda3\lib\site-packages\pymysql\cursors.py:170: Warning: (1366, "Incorrect string value: '\\xD6\\xD0\\xB9\\xFA\\xB1\\xEA...' for column 'VARIABLE_VALUE' at row 518")
  result = self._query(query)
复制代码

pandas + openpyxl 导出 Excel

pd.DataFrame 的 to_excel 方法可以导出数据到 Excel 指定 sheet ，但无法进一步调整格式：

第一个参数 excel_writer 设置 Excel 文件路径或已存在的 pd.ExcelWriter ；

第二个参数 sheet_name 设置 sheet 名称，默认 'Sheet1' ；

第三个参数 na_rep 设置缺失值填充，默认 '' ；

第六个参数 header 设置是否写入列标签，默认 True ；

第七个参数 index 设置是否写入行标签，默认 True 。

df.to_excel('D:\\test.xlsx','data',index=False)
复制代码

openpyxl 中的 Excel 对象主要有三个层次：

最顶层容器是 Workbook ，然后是 Worksheet ，最后是 Cell 。

openpyxl.load_workbook(file_path) 方法根据 Excel 文件路径加载 Workbook 对象， Workbook[sheet_name] 索引器可以根据名称获取 Worksheet ， Worksheet 的名称就是打开 Excel 文件时显示在下方的标签页名称，

Worksheet[cell_name] 索引器可以根据名称获取 Cell ， Cell 的名称是列标签与行号的组合，形如 'AB25' ，使用过 Excel 的人应该都比较熟悉。

Worksheet 还可通过 rows 或 columns 获取以行或列组织的 Cell 集合的迭代器，可用于行或列的遍历； row_dimensions[index] 或 column_dimensions[index] 获取用于设置基于行或列的属性的 RowDimension 或 ColumnDimension 对象，例如设置行列宽。

Cell 的属性：

value 属性存放单元格的值；

font 属性设置字体， Font 类型；

border 属性设置边框线， Border 类型；

alignment 属性设置对齐方式， Alignment 类型；

fill 属性设置填充， PatternFill 类型；

hyperlink 属性设置超链接， str 类型，格式形如 '#sheet!A1' 。

RowDimension 或 ColumnDimension 的共通属性：

height 或 width 属性设置行列宽；

hidden 属性设置是否隐藏；

font 、 border 、 alignment 、 fill 等属性的设置和 Cell 是一样。

color 的设置可以使用十六进制表示的 RGB 或 RGBA 字符串，例如 'FF000000' ，也可以使用 openpyxl.styles.colors 下预设的一些颜色配置。

from openpyxl import load_workbook
from openpyxl.styles import Font,Border,Side,Alignment,PatternFill,colors

wb=load_workbook('C:\\Users\\hp\\Desktop\\local_world.xlsx') 

ws=wb['Index']

print(ws['A1'].value)

ws['A1'].border=Border(
        left=Side(style='medium',color='FF000000'),
        right=Side(style='medium',color='FF000000'),
        top=Side(style='medium',color='FF000000'),
        bottom=Side(style='medium',color='FF000000')
        )

ws['A1'].font=Font(underline='single',color=colors.BLUE)

ws['A1'].alignment=Alignment(horizontal='center')

ws['A1'].fill=PatternFill(fill_type='solid',start_color='E6E6E6')

ws['A1'].hyperlink='#Table1!B2'

ws.column_dimensions['A'].width=40.0
复制代码