手写了个数据库工具

背景

现在服务的这家企业是一个传统企业，在信息系统建设的项目里面对交付的文件材料要求都比较多。举个简单的例子，在互联网数据库模型设计就直接在数据库进行了，最终交付的生产数据库的表里面信息都有，不会在单独存储一份word的文件。但是在这个企业需要将这些设计都写到word文档里，然后提交项目PMO管理系统进行项目验收、结项。

在微服务架构的项目里面，数据库会很多，而再B端信息系统里面，表会很多并且都是大宽表。如果采用手动的方式处理这个事情，会是非常耗费人力的。

于是我利用专业技能写了一个脚本，来解放上劳动力。

代码

# 导出数据库设计

#

import pymysql

from datetime import datetime

import xlsxwriter

today = datetime.now().strftime("%Y-%m-%d")

def all_tables(cursor, schema_name):

print(schema_name)

workbook = xlsxwriter.Workbook(f"../output/{today}_{schema_name}.xlsx")

sql_tables = f"SELECT table_name, table_comment FROM information_schema.tables WHERE table_schema = '{schema_name}';"

cursor.execute(sql_tables)

result = cursor.fetchall()

for table in result:

table_name = table[0]

print(table_name)

worksheet = workbook.add_worksheet(table_name if len(table_name) < 30 else table_name[:30])

worksheet.write(0, 0, table_name)

worksheet.write(0, 1, table[1])

table_design(cursor, schema_name, table_name, worksheet)

pass

workbook.close()

def table_design(cursor, target_schema, table_name, worksheet):

sql_columns = f"""

SELECT column_name, data_type, column_key, is_nullable, column_comment

FROM information_schema.columns

WHERE table_schema = '{target_schema}' AND table_name = '{table_name}';

"""

cursor.execute(sql_columns)

columns = cursor.fetchall()

print(f"表名：{table_name}")

index = 1

worksheet.write(index, 0, '列名')

worksheet.write(index, 1, '数据类型')

worksheet.write(index, 2, '键类型')

worksheet.write(index, 3, '是否可为空')

worksheet.write(index, 4, '注释')

for column in columns:

index += 1

print(f" 列名：{column[0]}, 数据类型：{column[1]}, 键类型：{column[2]}, 是否可为空：{column[3]}, 注释：{column[4]}")

worksheet.write(index, 0, column[0])

worksheet.write(index, 1, column[1])

worksheet.write(index, 2, column[2])

worksheet.write(index, 3, column[3])

worksheet.write(index, 4, column[4])

if __name__ == '__main__':

print("start")

connection = pymysql.connect(

host='127.0.0.1',

port=3306,

user='root',

password='root',

database='demo',

)

print("connected")

cursor = connection.cursor()

# query database

sql = 'SELECT schema_name FROM information_schema.schemata'

cursor.execute(sql)

result = cursor.fetchall()

for row in result:

schema_name = row[0]

if schema_name == 'information_schema' or schema_name == 'performance_schema':

continue

all_tables(cursor, schema_name)

cursor.close()

connection.close()