通过SSH隧道连接MySQL时出现“MySQL Connection not available”问题

我之前也碰到过类似的问题，既然你能用DBeaver正常连接，说明服务器端的配置是没问题的，咱们从代码和配置细节入手排查：

1. 先捕获详细错误信息，定位问题根源 #

你当前的代码没有异常处理，只得到模糊的“MySQL Connection not available”提示，先给代码加个异常捕获，看看具体是认证失败、数据库不存在还是其他问题：

import mysql.connector
import sshtunnel

try:
    with sshtunnel.SSHTunnelForwarder(
        ('server.web-hosting.com', 21098),
        ssh_username = 'ssh_username',
        ssh_password = 'ssh_pass!23',
        remote_bind_address = ('127.0.0.1', 3306)
    ) as tunnel:
        try:
            connection = mysql.connector.MySQLConnection(
                user = 'db_user',
                password = 'db_pass',
                host = '127.0.0.1',
                port = tunnel.local_bind_port,
                database = 'demo',
            )
            print("数据库连接成功！")
            # 后续查询逻辑...
        except mysql.connector.Error as err:
            print(f"详细数据库错误: {err}")
except sshtunnel.SSHTunnelForwarderError as tunnel_err:
    print(f"SSH隧道错误: {tunnel_err}")

2. 核对DBeaver与代码的配置一致性 #

把DBeaver里的配置和代码逐一对比：

• SSH配置：确认SSH主机、端口、用户名、密码完全一致，比如DBeaver是否用了密钥登录而代码用了密码？如果是密钥，代码需要添加ssh_pkey='/path/to/your/private_key'参数替换ssh_password。
• 数据库配置：DB用户名、密码、数据库名（注意大小写！有些共享主机的数据库名是带前缀的，比如karvy_demo而不是demo），这些要和DBeaver里的完全匹配。

3. 验证SSH隧道是否正常建立 #

有时候隧道建立需要一点时间，或者端口转发有问题，你可以：

• 在代码里添加打印本地端口的语句：print(f"本地绑定端口: {tunnel.local_bind_port}")，然后用这个端口在本地用mysql客户端测试连接：mysql -h 127.0.0.1 -P [本地端口] -u db_user -p，看看能不能连上。
• 或者用命令行手动建立隧道测试：ssh -L 3307:127.0.0.1:3306 ssh_username@server.web-hosting.com -p 21098，然后用DBeaver连127.0.0.1:3307，如果能连，说明SSH配置没问题，问题在代码逻辑。

4. 给隧道添加启动延迟（针对网络较慢的情况） #

共享主机的网络可能有延迟，隧道还没完全建立就尝试连接数据库会失败，给代码加个短延迟：

import time

# ... 隧道建立后
time.sleep(1) # 等待1秒让隧道稳定
connection = mysql.connector.MySQLConnection(...)

修复后的完整示例代码 #

结合上面的优化，完整代码如下：

import mysql.connector
import sshtunnel
import time

try:
    with sshtunnel.SSHTunnelForwarder(
        ('server.web-hosting.com', 21098),
        ssh_username='ssh_username',
        ssh_password='ssh_pass!23',
        remote_bind_address=('127.0.0.1', 3306)
    ) as tunnel:
        print(f"SSH隧道已建立，本地端口: {tunnel.local_bind_port}")
        time.sleep(1)
        
        try:
            connection = mysql.connector.MySQLConnection(
                user='db_user',
                password='db_pass',
                host='127.0.0.1',
                port=tunnel.local_bind_port,
                database='demo',
            )
            
            if connection.is_connected():
                print("数据库连接成功！")
                mycursor = connection.cursor()
                query = "SELECT * FROM sample_table"
                mycursor.execute(query)
                results = mycursor.fetchall()
                
                print(f"查询到 {len(results)} 条数据")
                # 可以循环打印结果
                for row in results:
                    print(row)
                
                # 清理资源
                mycursor.close()
                connection.close()
        except mysql.connector.Error as db_err:
            print(f"数据库连接失败: {db_err}")
except sshtunnel.SSHTunnelForwarderError as tunnel_err:
    print(f"SSH隧道建立失败: {tunnel_err}")
except Exception as e:
    print(f"其他未知错误: {e}")

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