最后一次更新日期：2019-5-28
分享一段简短的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'
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_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
关于拓展
(1). 表和列信息查询内容的增加只需要修改相应的sql就行了，但原有的TABLE_NAME列不能删除；
(2). 增加新的数据库类型支持需要拓展三处：
_create_engine方法中增加新类型的默认数据库和默认驱动设置；
query_schema_databases方法中增加新类型的数据库列表查询支持，返回数据库名列表；
query_schema_tables方法中增加新类型的表和列定义查询支持，返回两个DataFrame，注意，两个查询必须要包含可作为唯一标识的表名字段TABLE_NAME，另外同义的列保持列名统一，这样可以共用Excel文档的列宽设置；
(3). 预定义的一些Excel文档样式可在构造函数中修改，不想动默认设置的话也可以创建对象后单独从属性中修改，这些样式的用处都已经固定，更复杂的样式调整需要修改schema_to_excel方法。
