生产环境中，为了安全起见，大多数的数据库是无法在本地直接访问的，需要先连接跳板机，然后通过跳板机访问。创业天下测试数据库也是采用这种方法连接数据的，那么我们应该怎么通过Python去连接数据库呢，代码如下：
import pymysql
from sshtunnel import SSHTunnelForwarder
from common.log import *
from config.globalparameter import private_key_file, SqlConfig
class MysqlDB(object):
def __init__(self):
self.server = None
self.conn = None
self.cur = None
def start_server(self):
"""
创建连接对象
:return:
"""
self.server = SSHTunnelForwarder(
(SqlConfig["sshIP"], SqlConfig["sshPort"]), # 跳板机的配置
ssh_pkey=private_key_file,
ssh_username=SqlConfig["sshUserName"],
remote_bind_address=(SqlConfig["dbHost"], SqlConfig["dbPort"])) # 数据库的配置
self.server.start() # 连接跳板机服务
self.conn = pymysql.connect(host='127.0.0.1', # 此处必须是127.0.0.1
port=self.server.local_bind_port,
user=SqlConfig["dbUser"],
passwd=SqlConfig["dbPasswd"],
db=SqlConfig["dbName"])
self.cur = self.conn.cursor(cursor=pymysql.cursors.DictCursor) # 创建游标对象，指定数据类型为字典，将打印key,value
def select(self, sql):
"""
执行查询sql，执行成功返回结果,否则返回False，
:param sql: 执行的sql语句
:return:
"""
self.start_server()
try:
count = self.cur.execute(sql) # 执行sql语句
if count > 0:
data_db = self.cur.fetchall() # 获取数据
logger.info("Select succeeded！")
else:
logger.info("Select empty！")
data_db = False
except pymysql.Error as e:
data_db = False
logger.error("Select error:%s" % e)
finally:
self.cur.close()
self.conn.close()
self.server.close()
logger.info("All server closed")
return data_db
if __name__ == '__main__':
sql = "select user_name from `cytx_user_info` where telephone = 15229067460"
conn = MysqlDB()
value = conn.select(sql)[0]["user_name"]
print(value)
说明：
1：因为跳板机是通过密钥连接的，所以需要ssh_pkey参数，值是密钥的路径，如果需要通过密码连接，将该参数换成ssh_password即可
2：SSHTunnelForwarder方法返回的server对象必须调用start()方法后才可以正常使用
3：在连接MySQL时，connect()方法的参数中的host必须为127.0.0.1
4：导入的private_key_file,和SqlConfig分别是密钥存放的路径及数据库连接的配置信息
5：执行完sql语句后使用finally来关闭连接服务
6：导入sshtunnel 模块如果报下面的错误，请安装这个工具：链接：/s/1cWfjEsCpEFCJ42bLAGt9xg
提取码：ABCD ，因为这个模块部分代码需要c++编译环境
7：如果还是导入失败，尝试升级下你的pip版本：python -m pip install --upgrade pip
