Python 连接 mysql 可以借助 sqlalchemy 模块或者 pymsql 模块。

sqlalchemy

• 安装

pip install SQLAlchemy -i https://pypi.doubanio.com/simple

• 实际操作

from sqlalchemy import create_engine,text
from sshtunnel import SSHTunnelForwarder


def connect_db_with_engine_no_ssh():
    """
    连接数据库不依赖ssh通道
    连接参数：{0}:{1}@{2}:{3}/{4} 用户名:密码:主机名或IP地址:端口/数据库名
    :return:
    """
    try:
        engine = create_engine(
            'mysql+pymysql://{0}:{1}@{2}:{3}/{4}?charset=utf8'.format('username', 'password', 'ip', 'port',
                                                                      'database_name'), encoding='utf-8')
        sql = "show tables"
        df = pd.read_sql(text(sql), con=engine.connect())
        #engine.execute('show tables')
        print("df")
        print('不使用SSH连接数据库成功')
    except Exception as ex:
        print('数据库连接失败{}'.format(ex))
        raise ex


def connect_db_with_engine_need_ssh():
    """
    连接数据库依赖ssh通道
    连接参数：{0}:{1}@{2}:{3}/{4} 用户名:密码:主机名或IP地址:端口/数据库名
    :return:
    """
    try:
        server = SSHTunnelForwarder(
            ssh_address_or_host=('ssh_ip', 'ssh_port'),
            ssh_username='ssh_username',
            ssh_password='ssh_password',
            remote_bind_address=('mysql_ip', 'mysql_port'))
        server.start()
        engine = create_engine(
            'mysql+pymysql://{0}:{1}@{2}:{3}/{4}?charset=utf8'.format('mysql_username', 'mysql_password', '127.0.0.1',
                                                                      str(server.local_bind_port), 'mysql_database_name'),
            encoding='utf-8')
        engine.execute('show tables')
        print('使用SSH连接数据库成功')
    except Exception as ex:
        print('数据库连接失败{}'.format(ex))
        raise ex

pymsql

• 安装

pip install PyMySQL

• 实际操作

import pymysql
def connect_db_with_pymysql():
    try:
        db = pymysql.connect(host="ip", user="sername", password="password", port="port", db="database_name", charset="utf8")
        cursor = db.cursor()
        cursor.execute("SELECT VERSION()")
        print('连接数据库成功')
    except Exception as ex:
        print('数据库连接失败{}'.format(ex))
        raise ex