python 导出 MySQL 表结构、视图、存储过程和函数

博主： zmx
发布时间：2024 年 03 月 29 日
355 次浏览
暂无评论
5486字数
分类： MySQL Python

# 引言

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

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

# 一个一个整理不得累坏

## 工具类

```python
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
```

## 主程序

```python
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)
```

# 结语

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

![image.png](https://www.zunmx.top/usr/uploads/2024/03/2676050945.png)

如果觉得我的文章对你有用，请随意赞赏

发表评论取消回复
使用cookie技术保留您的个人信息以便您下次快速评论，继续评论表示您已同意该条款了昂。

评论 *

私密评论

名称 *

🎲

邮箱 *

地址

zunmx.fans
我 www.hao123.com实名申请友链OωO
zunmx.fans
作者小小骗人，我登陆了也看不到( ง ᵒ̌皿ᵒ̌)ง⁼³₌₃
zunmx.fans
please many update 算法 (´இ皿இ｀)boke
zunmx.fans
df1/df2 df1+df2 转置，累加----*。。。...
zunmx.fans
为何无人评论？OωO

python 导出 MySQL 表结构、视图、存储过程和函数

zmx • 2024 年 03 月 29 日

# 引言

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

# 一个一个整理不得累坏

## 工具类

```python
import pymysql

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

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) '

## 主程序

```python
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_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

# 结语

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

![image.png](https://www.zunmx.top/usr/uploads/2024/03/2676050945.png)

python 导出 MySQL 表结构、视图、存储过程和函数

发表评论取消回复
使用cookie技术保留您的个人信息以便您下次快速评论，继续评论表示您已同意该条款了昂。

Typecho 插件开发 ZSecurity

个人空间使用规则[拟定] 23/3/9修订

企业微信-摸鱼必备，避免离开状态

基于Lively wallpaper的壁纸

中公教育CTS视频下载

解决potplayer dsp列表为空

构建私密的网络传输环境

CentOS 防火墙操作

企业微信-摸鱼必备，避免离开状态

Android Studio 中模拟器卡Loading Devices

python 导出 MySQL 表结构、视图、存储过程和函数

发表评论 取消回复 使用cookie技术保留您的个人信息以便您下次快速评论，继续评论表示您已同意该条款了昂。

python 导出 MySQL 表结构、视图、存储过程和函数

发表评论取消回复
使用cookie技术保留您的个人信息以便您下次快速评论，继续评论表示您已同意该条款了昂。