首先需要一个模块sshtunnel，如果没有直接pip install sshtunnel

其实连个连接方式非常像：

pymysql连接方式：

import pymysqlfrom sshtunnel import SSHTunnelForwarderssh_host = ""  # 堡垒机ip地址或主机名ssh_port = 22  # 堡垒机连接mysql服务器的端口号，一般都是22，必须是数字ssh_user = ""  # 这是你在堡垒机上的用户名ssh_password = ""  # 这是你在堡垒机上的用户密码mysql_host = "localhost"  # 这是你mysql服务器的主机名或ip地址mysql_port = 3306  # 这是你mysql服务器上的端口，3306，mysql就是3306，必须是数字mysql_user = ""  # 这是你mysql数据库上的用户名mysql_password = ""  # 这是你mysql数据库的密码mysql_db = ""  # mysql服务器上的数据库名with SSHTunnelForwarder(        (ssh_host, ssh_port),        ssh_username=ssh_user,        ssh_password=ssh_password,        remote_bind_address=(mysql_host, mysql_port)) as server:    conn = pymysql.connect(host=mysql_host,                           port=server.local_bind_port,                           user=mysql_user,                           passwd=mysql_password,                           db=mysql_db)    cursor = conn.cursor()    cursor.execute("select * from user")    row_1 = cursor.fetchone()    print(row_1, type(row_1))    # 获取前n行数据    # row_2 = cursor.fetchmany(3)    # 获取所有数据    # row_3 = cursor.fetchall()    conn.commit()    cursor.close()    conn.close()

如果给pymysql加一个事务处理，可以看这里

sqlalchemy连接方式：

from sshtunnel import SSHTunnelForwarderfrom sqlalchemy import Column, String, Integer, create_engine, eventfrom sqlalchemy.orm import sessionmakerfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.exc import DisconnectionErrorssh_host = ""  # 堡垒机ip地址或主机名ssh_port = 22  # 堡垒机连接mysql服务器的端口号，一般都是22，必须是数字ssh_user = ""  # 这是你在堡垒机上的用户名ssh_password = ""  # 这是你在堡垒机上的用户密码mysql_host = ""  # 这是你mysql服务器的主机名或ip地址mysql_port = 3306  # 这是你mysql服务器上的端口，3306，mysql就是3306，必须是数字mysql_user = ""  # 这是你mysql数据库上的用户名mysql_password = ""  # 这是你mysql数据库的密码mysql_db = "" # mysql服务器上的数据库名Base = declarative_base()class Phones(Base):    __tablename__ = 'phones'    id = Column(Integer, primary_key=True)    name = Column(String(32))def checkout_listener(dbapi_con, con_record, con_proxy):    try:        try:            dbapi_con.ping(False)        except TypeError:            dbapi_con.ping()    except dbapi_con.OperationalError as exc:        if exc.args[0] in (2006, 2013, 2014, 2045, 2055):            raise DisconnectionError()        else:            raisewith SSHTunnelForwarder(        (ssh_host, ssh_port),        ssh_username=ssh_user,        ssh_password=ssh_password,        remote_bind_address=(mysql_host, mysql_port)) as server:    # server.start()  # ssh通道服务启动,用了with语句会自己启动    local_port = str(server.local_bind_port)    engine = create_engine(        'mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8'.format(mysql_user, mysql_password, '127.0.0.1', local_port,mysql_db),        pool_size=100,        pool_recycle=3600)    event.listen(engine, 'checkout', checkout_listener)  # 防止报连接池相关的错误    Base.metadata.create_all(engine)  # 检测文件中所有继承了Base类的类,在mysqld中建立所有的表,类就是表    Session = sessionmaker(bind=engine)    session = Session()