quality_tools/lib/mysql_tools.py

import mysql.connector
class MysqlUtil:
    @staticmethod
    def connect_to_mysql(host,port,user,password,database):
        # 创建数据库连接
        connection = mysql.connector.connect(
            host=host,  # 数据库主机地址
            user=user,  # 数据库用户名
            port=port,
            password=password,  # 数据库密码
            database=database  # 数据库名称
        )
        return  connection

    @staticmethod
    def execute_sql(connection,query,params):
        if connection.is_connected():
            print('Connected to MySQL database')
            # 创建一个cursor对象，用于执行SQL语句
            cursor = connection.cursor()
            # 执行SQL查询
            cursor.execute(query,params)
            mysql_count = cursor.fetchone()[0]
            cursor.close()
            connection.close()
            print('MySQL connection is closed')
            return mysql_count