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

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

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

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

效果图：

完整代码：

# -*- coding: utf-8 -*-import pandas as pdimport sqlalchemy as sqlafrom openpyxl import load_workbookfrom openpyxl.styles import Font,colors,Border,Side,Alignment,PatternFill   class SqlDataDictionary:    '''\n    Class: SQL数据字典        Parameters    ----------    dbtype: 数据库类型，仅支持'mysql'和'mssql'    username: 用户名    password: 密码    server: 服务器地址    database: 数据库名，不设置该参数时会连接默认数据库，              'mysql'默认'sys'，'mssql'默认'master'    charset: 字符集，默认'utf8'    driver: 驱动名，不设置该参数时会使用默认驱动，            'mysql'默认'pymysql'，'mssql'默认'pymssql'    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',driver=None,engine=None):                #生成engine        if type(engine)==type(None):            engine=self._create_engine(dbtype,driver,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,driver,username,password,server,database,charset):        #设置默认数据库和驱动        if dbtype=='mysql':            if type(database)==type(None): database='sys'            if type(driver)==type(None): driver='pymysql'        elif dbtype=='mssql':            if type(database)==type(None): database='master'            if type(driver)==type(None): driver='pymssql'        #此处可拓展其他数据库类型        else:            raise Exception('unsupported dbtype')                        engine=sqla.create_engine(                '{}+{}://{}:{}@{}/{}?charset={}'                .format(dbtype,driver,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')复制代码

说明：

该小工具主要有三部分：

从数据库系统视图中查询表和字段定义；

通过sqlalchemy和pandas执行查询sql并将结果以DataFrame返回；

通过pandas导出数据至Excel文件，再使用openpyxl调整Excel的格式。

查询表和字段定义

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

show databases;use db_nameselect * 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 pdresult=pd.read_sql('select * from test',engine)复制代码

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

import sqlalchemy as sqlaengine=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_workbookfrom openpyxl.styles import Font,Border,Side,Alignment,PatternFill,colorswb=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复制代码

关于拓展

(1). 表和列信息查询内容的增加只需要修改相应的sql就行了，但原有的TABLE_NAME列不能删除； (2). 增加新的数据库类型支持需要拓展三处：

_create_engine方法中增加新类型的默认数据库和默认驱动设置；

query_schema_databases方法中增加新类型的数据库列表查询支持，返回数据库名列表；

query_schema_tables方法中增加新类型的表和列定义查询支持，返回两个DataFrame，注意，两个查询必须要包含可作为唯一标识的表名字段TABLE_NAME，另外同义的列保持列名统一，这样可以共用Excel文档的列宽设置；

(3). 预定义的一些Excel文档样式可在构造函数中修改，不想动默认设置的话也可以创建对象后单独从属性中修改，这些样式的用处都已经固定，更复杂的样式调整需要修改schema_to_excel方法。