Python通过跳板机访问数据库的方法

from sshtunnel import SSHTunnelForwarder
server = SSHTunnelForwarder(
        (host_jump, int(port_jump)),  # 跳板机的配置
        ssh_pkey=ssh_pk_jump, 
        #密钥的访问方式，如果是密码 ssh_passWord=？？？
        ssh_username=user_name_jump,
        remote_bind_address=(host_Mysql, int(port_mysql)))  # 数据库服务器的配置
        local_bind_address=(local_ip, local_port)  # 开启本地转发端口
        ）
	server.start()
	server.close()

from sshtunnel import SSHTunnelForwarder
with SSHTunnelForwarder(
        (host_jump, int(port_jump)),  # 跳板机的配置
        ssh_pkey=ssh_pk_jump, 
        #密钥的访问方式，如果是密码 ssh_password=？？？
        ssh_username=user_name_jump,
        remote_bind_address=(host_mysql, int(port_mysql)))  # 数据库服务器的配置
        local_bind_address=(local_ip, local_port)  # 开启本地转发端口
        ）as server
	server.start()

with SSHTunnelForwarder(
        (host_jump, int(port_jump)),  # 跳板机的配置
        ssh_pkey=ssh_pk_jump, 
        #密钥的访问方式，如果是密码 ssh_password=？？？
        ssh_username=user_name_jump,
        remote_bind_address=(host_mysql, int(port_mysql)))  # 数据库服务器的配置
        local_bind_address=(local_ip, local_port)  # 开启本地转发端口
        ）as server
	server.start()
    conn = mysql.connector.connect(**{'host': '127.0.0.1',
                              'port': server.local_bind_port,
                              'user': dbcfg["user"],
                              'password': dbcfg["password"],
                              'db': dbcfg["db"],
                              'charset': dbcfg["charset"],
                              'collation': dbcfg["collation"],
                              'use_unicode': dbcfg["use_unicode"],
                              'use_pure': True})
        cursor = conn.cursor(dictionary=True)
        cursor.execute(sql, params)
        rows = cursor.fetchall()
        # conn.commit()
        cursor.close()
        conn.close()

def get_slave_conn_byProxy(dbcfg):
    server = SSHTunnelForwarder(
        eval(dbcfg["addr"]),
        ssh_username=dbcfg["ssh_username"],
        ssh_pkey=os.getenv("HOME") + "/.ssh/id_rsa",
        remote_bind_address=eval(dbcfg["remote_bind_address"]),
        local_bind_address=eval(dbcfg["local_bind_address"])  # 开启本地转发端口
    )
    server.start()
    conn = get_db_connection({'host': '127.0.0.1',
                              'port': server.local_bind_port,
                              'user': dbcfg["user"],
                              'password': dbcfg["password"],
                              'db': dbcfg["db"],
                              'charset': dbcfg["charset"],
                              'collation': dbcfg["collation"],
                              'use_unicode': dbcfg["use_unicode"],
                              'use_pure': True})
    return server, conn

def get_db_connection(dbcfg):
    # logger.info(f"============ debug db config: {dbcfg}")
    conn = mysql.connector.connect(**dbcfg)
    return conn

# 通过输入的连接参数判断是否需要通过跳板机的方式
def get_slave_conn(dbcfg):
    if "addr" in dbcfg:
        return get_slave_conn_byProxy(dbcfg)
    else:
        conn = get_db_connection(dbcfg)
        return None, conn
       
def read_slave_db_data(sql, conn, params=None, use_pdf=True):
        cursor = conn.cursor(dictionary=True)
        cursor.execute(sql, params)
        rows = cursor.fetchall()
        # conn.commit()
        cursor.close()
        if use_pdf:
            return pd.DataFrame(rows)
        else:
            return rows