快捷導航

Python腳本實現(xiàn)一鍵執(zhí)行MySQL與達夢數(shù)據(jù)庫的SQL

更新時間：2025年08月25日 09:38:07 作者：碼頭薯條整點

這篇文章主要為大家詳細介紹了如何使用Python 腳本 + PyInstaller 打包成可執(zhí)行文件,一鍵搞定 MySQL 和達夢（DM）數(shù)據(jù)庫的 SQL 執(zhí)行,希望對大家有所幫助

這時候，如果能有一個小巧的工具，直接在命令行里執(zhí)行 SQL，是不是會方便很多？今天這篇文章，就帶大家看看如何用 Python 腳本 + PyInstaller 打包成可執(zhí)行文件，一鍵搞定 MySQL 和達夢（DM）數(shù)據(jù)庫的 SQL 執(zhí)行。

環(huán)境準備

首先你需要一個 Linux 環(huán)境，并且提前安裝好 conda。本文示例在 x86 和 ARM 架構(gòu)都測試過。

創(chuàng)建 Python 環(huán)境：

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 腳本實現(xiàn)

核心思路很簡單：

用 argparse 解析命令行參數(shù)
根據(jù)選擇的數(shù)據(jù)庫類型建立連接
支持執(zhí)行 單條 SQL 或 SQL 文件
打印結(jié)果

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

# 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("執(zhí)行結(jié)果：")
        print(" | ".join(columns))
        print("-" * 50)
        for row in rows:
            print(" | ".join(str(col) for col in row))
    else:
        print("SQL 執(zhí)行成功，沒有返回結(jié)果。")

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"執(zhí)行文件中的 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 數(shù)據(jù)庫！")
        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) 數(shù)據(jù)庫！")
        else:
            raise ValueError("不支持的數(shù)據(jù)庫類型")

        cursor = connection.cursor()
        if sql:
            print(f"執(zhí)行 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} 數(shù)據(jù)庫。錯誤信息:", e)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="測試 MySQL 或 達夢(DM) 數(shù)據(jù)庫連接并執(zhí)行 SQL")
    subparsers = parser.add_subparsers(dest="db_type", required=True, help="選擇數(shù)據(jù)庫類型：mysql 或 dm")

    # MySQL
    mysql_parser = subparsers.add_parser("mysql", help="測試 MySQL 連接并執(zhí)行 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) 連接并執(zhí)行 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)

參數(shù)說明

參數(shù)名	是否必填	說明	默認值	示例值
--host	?	數(shù)據(jù)庫主機名或 IP	-	192.168.1.13
--port	?	數(shù)據(jù)庫端口	3306 / 5236	33066
--dbname	?	MySQL 庫名 / DM Schema	-	testdb
--user	?	用戶名	-	root / SYSDBA
--password	?	密碼	-	123456
--sql	?	要執(zhí)行的 SQL	-	"show databases"
--sqlfile	?	SQL 文件路徑	-	./init.sql

注意事項：

--sql 與 --sqlfile 不能同時用。
在 DM 數(shù)據(jù)庫中，--dbname 代表 Schema 名。
在 MySQL 中，--dbname 可選，不寫時可以直接查詢?nèi)謹?shù)據(jù)。

示例用法

# 查詢 MySQL 所有數(shù)據(jù)庫
python exec_sql.py mysql --host 192.168.1.13 --port 33066 --user root --password 123456 --sql "show databases"

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

# 查詢達夢數(shù)據(jù)庫表
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"

# 執(zhí)行達夢 SQL 文件
python exec_sql.py dm --host 192.168.11.72 --port 5236 --dbname DATAMASK_WEB --user SYSDBA --password SYSDBA --sqlfile ./init.sql

打包成可執(zhí)行文件

安裝 staticx，把 glibc 鏈接庫也打包進去，避免環(huán)境差異：

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 就是最終的可執(zhí)行文件。

驗證效果

運行：

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

只要能正常打印數(shù)據(jù)庫列表，就說明工具搞定啦

總結(jié)

通過這樣一個小工具，我們可以：

無需圖形化客戶端
跨平臺（x86 / ARM）
一鍵執(zhí)行 SQL 或 SQL 文件

非常適合在客戶環(huán)境下的應(yīng)急運維場景！

到此這篇關(guān)于Python腳本實現(xiàn)一鍵執(zhí)行MySQL與達夢數(shù)據(jù)庫的SQL的文章就介紹到這了,更多相關(guān)Python執(zhí)行MySQL與達夢數(shù)據(jù)庫內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: