基本数据配置

# 阿里云数据库地址
db_config = {
    "host": "121.196.***.***",
    "port": 22,
    "username": "root",
    "password": "********",
    "address": "***********.mysql.rds.aliyuncs.com",
    "address_port": 2019
}

# 本地连接数据库配置
local_db_config = {
    "host": "127.0.0.1",
    "username": "dbtestuser",
    "password": "dbtest123!",
    "db_name": "dbuseruser"
}

连接数据库

import MySQLdb
from config import db_config as config
from sshtunnel import SSHTunnelForwarder


class DBQuery:
    def __init__(self):
        self.server = SSHTunnelForwarder(
            (config.db_config["host"], config.db_config["port"]),   # B机器的配置
            ssh_username=config.db_config["username"],
            ssh_password=config.db_config["password"],
            remote_bind_address=(config.db_config["address"], config.db_config["address_port"])
        )
        self.server.start()
        self.conn = MySQLdb.connect(
            host=config.local_db_config["host"],
            port=self.server.local_bind_port,
            user=config.local_db_config["username"],
            passwd=config.local_db_config["password"],
            db=config.local_db_config["db_name"]
        )

    def db_execute(self, sql_str):
        """
        数据库执行
        :return:执行语句
        :return:执行结果
        """
        cursor = self.conn.cursor()
        # noinspection PyBroadException
        try:    # 查询操作
            print(sql_str)
            cursor.execute(sql_str)
            self.conn.commit()
            return cursor.fetchall()
        except Exception:       # 异常处理
            print("-------------数据库执行出错------------")
            self.conn.rollback()
        finally:    # 断开数据库
            self.conn.close()
            self.server.close()