如何在使用SSHTunnel连接远程MySQL时实现Flask DB迁移升级？

当然可以解决这个问题！我之前也遇到过类似的场景——用SSH隧道连接远程数据库时，Flask-Migrate因为没拿到正确的数据库URI而无法执行迁移。核心思路就是让flask db命令能感知到SSH隧道的存在，并使用隧道映射的本地端口来构建SQLALCHEMY_DATABASE_URI，下面是两种可靠的实现方式：

方法一：在应用初始化时绑定隧道与数据库URI #

直接修改你的Flask应用初始化文件（比如__init__.py），让应用启动（包括迁移命令执行时）先启动SSH隧道，再动态设置数据库连接配置。这样flask db命令加载应用时会自动触发隧道启动，拿到正确的本地端口：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
import sshtunnel

# 初始化Flask应用
app = Flask(__name__)

# 启动SSH隧道并配置数据库
try:
    sshtunnel.SSH_TIMEOUT = 5.0
    sshtunnel.TUNNEL_TIMEOUT = 5.0
    server = sshtunnel.SSHTunnelForwarder(
        ('ssh.pythonanywhere.com', 22),
        ssh_password="mypassword",
        ssh_username="myusername",
        remote_bind_address=('myname.mysql.pythonanywhere-services.com', 3306)
    )
    server.start()

    # 动态生成数据库URI，使用隧道映射的本地端口
    app.config['SQLALCHEMY_DATABASE_URI'] = (
        f'mysql+mysqldb://mynameb:dbpassword@127.0.0.1:{server.local_bind_port}/dbname'
    )
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

    # 初始化SQLAlchemy和Flask-Migrate
    db = SQLAlchemy(app)
    migrate = Migrate(app, db)

    # 确保应用关闭时关闭隧道，避免资源泄漏
    @app.teardown_appcontext
    def close_ssh_tunnel(exception=None):
        if server.is_active:
            server.stop()

except Exception as e:
    print(f"SSH隧道启动失败: {str(e)}")
    raise

配置完成后，直接运行flask db upgrade即可——命令会自动加载应用初始化逻辑，启动隧道并连接到远程数据库执行迁移。

方法二：创建独立的迁移脚本（更灵活） #

如果不想修改应用核心初始化逻辑，可以单独写一个脚本，专门处理带SSH隧道的迁移操作：

# migrate_with_ssh.py
import sshtunnel
from flask_migrate import upgrade
from your_app import app  # 替换成你的Flask应用实例的导入路径

def run_migration():
    # 启动SSH隧道
    sshtunnel.SSH_TIMEOUT = 5.0
    sshtunnel.TUNNEL_TIMEOUT = 5.0
    server = sshtunnel.SSHTunnelForwarder(
        ('ssh.pythonanywhere.com', 22),
        ssh_password="mypassword",
        ssh_username="myusername",
        remote_bind_address=('myname.mysql.pythonanywhere-services.com', 3306)
    )

    try:
        server.start()
        # 临时覆盖应用的数据库URI
        app.config['SQLALCHEMY_DATABASE_URI'] = (
            f'mysql+mysqldb://mynameb:dbpassword@127.0.0.1:{server.local_bind_port}/dbname'
        )
        # 在应用上下文内执行迁移
        with app.app_context():
            upgrade()
        print("数据库迁移执行成功")
    except Exception as e:
        print(f"迁移失败: {str(e)}")
    finally:
        # 无论成功失败，都关闭隧道
        if server.is_active:
            server.stop()

if __name__ == '__main__':
    run_migration()

运行时直接执行python migrate_with_ssh.py即可，不需要依赖flask db命令，适合需要单独控制迁移流程的场景。

注意事项 #

• 避免硬编码密码：实际生产环境中，建议用环境变量（比如os.getenv("SSH_PASSWORD")）来存储SSH和数据库密码，不要直接写在代码里。
• 异常处理：一定要加隧道启动失败的捕获逻辑，避免应用或迁移脚本崩溃导致资源泄漏。
• 端口冲突：如果本地3306端口被占用，可以在SSHTunnelForwarder里设置local_bind_address=('127.0.0.1', 0)让系统自动分配可用端口，脚本会自动获取server.local_bind_port。

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