引言

有个老项目要写文档，部门老大让我帮忙整理一下文档，说是全部用存储过程写的，就让我一个一个的导出来，分别保存成一个文件。
比如说库A里面有表B、视图C、存储过程D和函数E
然后要求格式为

A
|-TABLE
  |-B
|-VIEW
  |-C
|-PROCEDURE
  |-D
|-FUNCTION
  |-E

一个一个整理不得累坏

工具类

import pymysql

global connection
connection = None
user = 'root'
password = '********'
host = '172.*.*.*'
port = ********
database = '********'


def get_connection():
    global connection
    if connection is None:
        connection = pymysql.Connection(host=host, port=port, database=database, user=user, password=password)
    else:
        connection.ping(reconnect=True)
    return connection


def insert(datas: dict, database: str, table: str):
    k = []
    v = []
    for key in datas.keys():
        k.append(f'''`{key}`''')
        v.append(str(datas[key]))
    sql = f'''INSERT INTO {database}.{table}({','.join(k)}) VALUES("{'","'.join(v)}")'''
    execute(sql)


def query(sql):
    try:
        con = get_connection()
        cursor = con.cursor(cursor=pymysql.cursors.DictCursor)
        cursor.execute(sql)
        fetchall = cursor.fetchall()
        return fetchall
    except Exception as e:
        return e


def execute(sql):
    try:
        con = get_connection()
        cursor = con.cursor(cursor=pymysql.cursors.DictCursor)
        rst = cursor.execute(sql)
        con.commit()
        return rst
    except Exception as e:
        return e


def __get_type(v):
    if type(v) == str:
        return ' varchar(1024) '
    elif type(v) == int:
        return ' bigint '
    elif type(v) == float:
        return ' decimal(18,4) '


def dictkey_to_create_table(dct: dict, database: str, table: str):
    sql = f'CREATE TABLE `{database}`.`{table}`('
    for key in dct.keys():
        sql += f'''`{key}` {__get_type(dct[key])} ,'''
    sql = sql[0:-1] + ")"
    return sql

主程序

import os.path

from utils import SQL


def get_table_list(db_name):
    sql = f"SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = '{db_name}' AND TABLE_TYPE = 'BASE TABLE'"
    query = SQL.query(sql)
    return query


def get_view_list(db_name):
    sql = f"SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = '{db_name}' AND TABLE_TYPE = 'VIEW'"
    query = SQL.query(sql)
    return query


def get_procedure_list(db_name):
    sql = f"SELECT ROUTINE_NAME FROM information_schema.ROUTINES WHERE ROUTINE_SCHEMA = '{db_name}' AND ROUTINE_TYPE = 'PROCEDURE'"
    query = SQL.query(sql)
    return query


def get_function_list(db_name):
    sql = f"SELECT ROUTINE_NAME FROM information_schema.ROUTINES WHERE ROUTINE_SCHEMA = '{db_name}' AND ROUTINE_TYPE = 'FUNCTION'"
    query = SQL.query(sql)
    return query


def get_table_struct(db_name, tb_name):
    sql = f"SHOW CREATE TABLE {db_name}.{tb_name}"
    query = SQL.query(sql)
    return query


def get_view_struct(db_name, tb_name):
    sql = f"SHOW CREATE VIEW {db_name}.{tb_name}"
    query = SQL.query(sql)
    return query


def get_procedure_struct(db_name, tb_name):
    sql = f"SHOW CREATE PROCEDURE {db_name}.{tb_name}"
    query = SQL.query(sql)
    return query


def get_function_struct(db_name, tb_name):
    sql = f"SHOW CREATE FUNCTION {db_name}.{tb_name}"
    query = SQL.query(sql)
    return query


def write_file(path, name, category, content):
    if not os.path.exists(os.path.join(path, category)):
        os.makedirs(os.path.join(path, category))
    with open(os.path.join(path, category, name + ".sql"), 'w', encoding='utf8') as f:
        f.write(content)
    print('[v]', 'export', category, path, name, 'Success')


def output_table_struct(db):
    table_list = get_table_list(db['Database'])
    for tb_name in table_list:
        struct = get_table_struct(db['Database'], tb_name['TABLE_NAME'])
        write_file(os.path.join('output', db['Database']), tb_name['TABLE_NAME'], 'table', struct[0]['Create Table'])


def output_view_struct(db):
    view_list = get_view_list(db['Database'])
    for view_name in view_list:
        struct = get_view_struct(db['Database'], view_name['TABLE_NAME'])
        write_file(os.path.join('output', db['Database']), view_name['TABLE_NAME'], 'view', struct[0]['Create View'])

def output_procedure_struct(db):
    procedure_list = get_procedure_list(db['Database'])
    for procedure_name in procedure_list:
        struct = get_procedure_struct(db['Database'], procedure_name['ROUTINE_NAME'])
        write_file(os.path.join('output', db['Database']), procedure_name['ROUTINE_NAME'], 'procedure', struct[0]['Create Procedure'])

def output_function_struct(db):
    function_list = get_function_list(db['Database'])
    for function_name in function_list:
        struct = get_function_struct(db['Database'], function_name['ROUTINE_NAME'])
        write_file(os.path.join('output', db['Database']), function_name['ROUTINE_NAME'], 'function', struct[0]['Create Function'])

if __name__ == '__main__':
    query = SQL.query("SHOW DATABASES")
    for db in query:
        if db['Database'].startswith('asset_'): # 排除非asset开头的库（这里需要根据实际情况进行修改）
            output_table_struct(db)
            output_view_struct(db)
            output_procedure_struct(db)
            output_function_struct(db)

结语

善用工具，事半功倍。
学会偷懒，快乐摸鱼。