在没有客户端的客户环境下，如何用 Python 一键执行 MySQL 与达梦数据库 SQL

在客户现场做运维时，大家是不是经常会遇到这种尴尬场景：

• 没有图形化的数据库客户端
• 自带的数据库终端又超级难用

这时候，如果能有一个小巧的工具，直接在命令行里执行 SQL，是不是会方便很多？今天这篇文章，就带大家看看如何用 Python 脚本 + PyInstaller 打包成可执行文件，一键搞定 MySQL 和达梦（DM）数据库的 SQL 执行。

---

环境准备

首先你需要一个 Linux 环境，并且提前安装好 conda。本文示例在 x86 和 ARM 架构都测试过。

创建 Python 环境：

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

安装依赖（目前仅支持 MySQL 和 DM，如需扩展可自行修改）：

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

---

Python 脚本实现

核心思路很简单：

1. 用 argparse 解析命令行参数
2. 根据选择的数据库类型建立连接
3. 支持执行 单条 SQL 或 SQL 文件
4. 打印结果

下面是完整的脚本（可直接使用）：

# exec_sql.py
import argparse
import pymysql
import dmPython
import os
from pymysql import OperationalError as MySQLError
from dmPython import Error as DMErrordef 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}")returnwith 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"] = dbnameconnection = 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")# MySQLmysql_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")# DMdm_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)

---

参数说明

参数名	是否必填	说明	默认值	示例值
--host	✅	数据库主机名或 IP	-	192.168.1.13
--port	❌	数据库端口	3306 / 5236	33066
--dbname	❌	MySQL 库名 / DM Schema	-	testdb
--user	✅	用户名	-	root / SYSDBA
--password	✅	密码	-	123456
--sql	❌	要执行的 SQL	-	"show databases"
--sqlfile	❌	SQL 文件路径	-	./init.sql

⚠️ 注意事项：

• --sql 与 --sqlfile 不能同时用。
• 在 DM 数据库中，--dbname 代表 Schema 名。
• 在 MySQL 中，--dbname 可选，不写时可以直接查询全局数据。

---

示例用法

# 查询 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

---

打包成可执行文件

安装 staticx，把 glibc 链接库也打包进去，避免环境差异：

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

这时 dist/exec_sql_static 就是最终的可执行文件。

---

验证效果

运行：

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

只要能正常打印数据库列表，就说明工具搞定啦

---

总结

通过这样一个小工具，我们可以：

• 无需图形化客户端
• 跨平台（x86 / ARM）
• 一键执行 SQL 或 SQL 文件

非常适合在客户环境下的应急运维场景！