SQL Workbench可连MySQL，如何用Python通过TCP/IP及SSH连接？

首先安装数据库连接依赖库，推荐两种常用库二选一：

# 方案1：使用mysql-connector-python
pip install mysql-connector-python

# 方案2：使用pymysql
pip install pymysql

基于mysql-connector-python的连接代码 #

把代码中的参数替换成你SQL Workbench里的实际配置：

import mysql.connector
from mysql.connector import Error

try:
    # 替换为你图中的数据库参数
    connection = mysql.connector.connect(
        host='你的数据库主机地址',  # 如192.168.xxx.xxx或域名
        port='你的数据库端口',    # 默认3306，以你实际设置为准
        user='你的数据库用户名',
        password='你的数据库密码',
        database='目标数据库名'  # 可选，直接连接到指定库
    )
    if connection.is_connected():
        db_info = connection.get_server_info()
        print(f"成功连接到MySQL服务器版本: {db_info}")
        cursor = connection.cursor()
        cursor.execute("SELECT DATABASE();")
        record = cursor.fetchone()
        print(f"当前连接的数据库: {record[0]}")

except Error as e:
    print(f"连接出错: {e}")
finally:
    if 'connection' in locals() and connection.is_connected():
        cursor.close()
        connection.close()
        print("数据库连接已关闭")

基于pymysql的连接代码 #

逻辑和上面一致，只是库的调用方式不同：

import pymysql

try:
    connection = pymysql.connect(
        host='你的数据库主机地址',
        port=3306,  # 注意这里是数字类型
        user='你的数据库用户名',
        password='你的数据库密码',
        database='目标数据库名'
    )
    with connection.cursor() as cursor:
        cursor.execute("SELECT VERSION();")
        version = cursor.fetchone()
        print(f"MySQL服务器版本: {version[0]}")

except pymysql.MySQLError as e:
    print(f"连接出错: {e}")
finally:
    connection.close()

这种方式需要先建立SSH隧道，将远程数据库端口转发到本地，再进行连接。先安装依赖库：

pip install sshtunnel mysql-connector-python

完整连接代码 #

替换代码中SSH和数据库的实际参数：

from sshtunnel import SSHTunnelForwarder
import mysql.connector
from mysql.connector import Error

# SSH服务器参数（替换为你图中的SSH配置）
ssh_host = 'SSH服务器地址'
ssh_port = 22  # 默认SSH端口，以实际设置为准
ssh_user = 'SSH登录用户名'
ssh_password = 'SSH登录密码'  # 如果用密钥登录，替换为ssh_pkey='./你的私钥文件路径'

# 远程数据库参数（注意是SSH内网侧的数据库地址，比如同服务器的话填127.0.0.1）
db_host = '远程数据库内网地址'
db_port = 3306
db_user = '数据库用户名'
db_password = '数据库密码'
db_name = '目标数据库名'

# 建立SSH隧道并连接数据库
with SSHTunnelForwarder(
    (ssh_host, ssh_port),
    ssh_username=ssh_user,
    ssh_password=ssh_password,
    remote_bind_address=(db_host, db_port)
) as tunnel:
    print(f"本地转发端口: {tunnel.local_bind_port}")
    try:
        connection = mysql.connector.connect(
            host='127.0.0.1',
            port=tunnel.local_bind_port,
            user=db_user,
            password=db_password,
            database=db_name
        )
        if connection.is_connected():
            print("成功通过SSH隧道连接到数据库")
            # 示例：执行SQL查询
            cursor = connection.cursor()
            cursor.execute("SELECT NOW();")
            result = cursor.fetchone()
            print(f"当前数据库时间: {result[0]}")

    except Error as e:
        print(f"连接出错: {e}")
    finally:
        if 'connection' in locals() and connection.is_connected():
            cursor.close()
            connection.close()
            print("数据库连接已关闭")

注意事项 #

• 所有参数必须严格匹配你的配置，尤其是主机地址、端口、用户名密码，不能有拼写错误
• 使用SSH密钥登录时，私钥文件权限需设置正确（比如Linux下设为600）
• 若连接失败，先检查防火墙是否开放对应端口、是否有IP白名单限制
• 确保Python环境能正常访问SSH服务器和数据库服务器的网络

内容的提问来源于stack exchange，提问作者Ashutosh Yadav