Python脚本实现一键执行MySQL与达梦数据库的SQL

conda create -n exec-sql python=3.10
conda activate exec-sql

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pymysql dmPython pyinstaller

# exec_sql.py
import argparse
import pymysql
import dmPython
import os
from pymysql import OperationalError as MySQLError
from dmPython import Error as DMError

def exec_and_print(cursor, sql):
    cursor.execute(sql)
    if cursor.description:
        columns = [col[0] for col in cursor.description]
        rows = cursor.fetchall()
        print("执行结果：")
        print(" | ".join(columns))
        print("-" * 50)
        for row in rows:
            print(" | ".join(str(col) for col in row))
    else:
        print("SQL 执行成功，没有返回结果。")

def execute_sql_from_file(cursor, sqlfile):
    if not os.path.exists(sqlfile):
        print(f"SQL 文件不存在: {sqlfile}")
        return
    with open(sqlfile, "r", encoding="utf-8") as f:
        sql_script = f.read()
    for stmt in sql_script.split(";"):
        stmt = stmt.strip()
        if stmt:
            print(f"执行文件中的 SQL: {stmt}")
            exec_and_print(cursor, stmt)

def test_db_connection(db_type, host, port, dbname, user, password, sql=None, sqlfile=None):
    try:
        if db_type == "mysql":
            conn_params = {"host": host, "port": port, "user": user, "password": password, "connect_timeout": 5}
            if dbname:
                conn_params["database"] = dbname
            connection = pymysql.connect(**conn_params)
            print("成功连接到 MySQL 数据库！")
        elif db_type == "dm":
            connection = dmPython.connect(user=user, password=password, server=host, port=port)
            if dbname:
                cursor = connection.cursor()
                cursor.execute(f"set schema {dbname}")
                cursor.close()
            print("成功连接到 达梦(DM) 数据库！")
        else:
            raise ValueError("不支持的数据库类型")

        cursor = connection.cursor()
        if sql:
            print(f"执行 SQL: {sql}")
            exec_and_print(cursor, sql)
        if sqlfile:
            execute_sql_from_file(cursor, sqlfile)
        cursor.close()
        connection.close()
    except (MySQLError, DMError) as e:
        db_name = "MySQL" if db_type == "mysql" else "达梦(DM)"
        print(f"无法连接到 {db_name} 数据库。错误信息:", e)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="测试 MySQL 或 达梦(DM) 数据库连接并执行 SQL")
    subparsers = parser.add_subparsers(dest="db_type", required=True, help="选择数据库类型：mysql 或 dm")

    # MySQL
    mysql_parser = subparsers.add_parser("mysql", help="测试 MySQL 连接并执行 SQL")
    mysql_parser.add_argument("--host", required=True)
    mysql_parser.add_argument("--port", type=int, default=3306)
    mysql_parser.add_argument("--dbname")
    mysql_parser.add_argument("--user", required=True)
    mysql_parser.add_argument("--password", required=True)
    mysql_parser.add_argument("--sql")
    mysql_parser.add_argument("--sqlfile")

    # DM
    dm_parser = subparsers.add_parser("dm", help="测试 达梦(DM) 连接并执行 SQL")
    dm_parser.add_argument("--host", required=True)
    dm_parser.add_argument("--port", type=int, default=5236)
    dm_parser.add_argument("--dbname")
    dm_parser.add_argument("--user", required=True)
    dm_parser.add_argument("--password", required=True)
    dm_parser.add_argument("--sql")
    dm_parser.add_argument("--sqlfile")

    args = parser.parse_args()
    test_db_connection(args.db_type, args.host, args.port, args.dbname, args.user, args.password, args.sql, args.sqlfile)

# 查询 MySQL 所有数据库
python exec_sql.py mysql --host 192.168.1.13 --port 33066 --user root --password 123456 --sql "show databases"

# 执行 SQL 文件
python exec_sql.py mysql --host 127.0.0.1 --port 3306 --dbname testdb --user root --password 123456 --sqlfile ./init.sql

# 查询达梦数据库表
python exec_sql.py dm --host 192.168.11.72 --port 5236 --dbname datamask_web --user SYSDBA --password SYSDBA --sql "select * from job_jdbc_datasource"

# 执行达梦 SQL 文件
python exec_sql.py dm --host 192.168.11.72 --port 5236 --dbname DATAMASK_WEB --user SYSDBA --password SYSDBA --sqlfile ./init.sql

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple staticx
sudo apt-get install patchelf musl-tools -y

pyinstaller -F exec_sql.py
staticx dist/exec_sql dist/exec_sql_static

./exec_sql_static mysql --host 192.168.1.13 --port 33066 --user root --password 123456 --sql "show databases"