导入Jpype第三方包，驱动jar包连接数据库。

前提准备

# 下载第三方包
pip install -i [https://pypi.mirrors.ustc.edu.cn/simple/] jpype1
# 导入第三方包
import jpype.imports
from jpype import dbapi2
# 加载jar包
jpype.startJVM(classpath='D:/Enviroment/Java/jar/mysql-connector-java-8.0.29/mysql-connector-java-8.0.29.jar')
# 数据库配置信息
HOST = '101.xx.xx.xx'
PORT = '3306'
DATABASE = 'db_test'
USERNAME = 'xxxx'
PASSWORD = 'xxxx'

构造 MySQL 类

class MySqlDB(object):
    def __init__(self, host=HOST, port=PORT, db=DATABASE, user=USERNAME, passwd=PASSWORD):
        self.conn_url = "jdbc:mysql://{host}:{port}/{db}?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false".format(host=host, port=port, db=db)
        self.driver_args = {"user": user, "password": passwd}
        self.driver = "com.mysql.cj.jdbc.Driver"

    def connect(self):
        self.conn = dbapi2.connect(self.conn_url, driver=self.driver, driver_args=self.driver_args)
        self.cursor = self.conn.cursor()

    def close(self):
        self.cursor.close()
        self.conn.close()

    def get_all(self, sql):
        """
        拿取所有的结果
        :param sql:
        :return:
        """
        self.cursor.execute(sql)
        result = self.cursor.fetchall()
        return result

    def get_one(self, sql):
        """
        拿取第一个结果
        :param sql:
        :return:
        """
        self.cursor.execute(sql)
        result = self.cursor.fetchone()
        return result

连接操作

查询数据库下的表

def get_tables():
    mysql_obj = MySqlDB()
    mysql_obj.connect()
    list = mysql_obj.get_all("show tables;")
    return list

查询数据库表的信息

def get_tableinfo(table):
    mysql_obj = MySqlDB()
    mysql_obj.connect()
    sql = """select * from {table};""".format(table=table)
    result = mysql_obj.get_all(sql)
    return result