text_to_vector/utils/mysql_helper.py

# -*- coding: utf-8 -*-
# @Time    : 2023/8/26 
# @Author  : lkj
# @description :
import time

from loguru import logger
import mysql.connector


class MysqlConn(object):
    def __init__(self,mysql_path):
        self.mysql_path = mysql_path
        self.cursor = ''
        self.mysql_conn = ''
        self.mysql_conn, self.cursor = self.link_mysql()

    def link_mysql(self,):
        """
        连接
        """
        try:
            mysql_conn = mysql.connector.connect(
                host=self.mysql_path.get('mysql_host'),
                port=self.mysql_path.get('mysql_port'),
                user=self.mysql_path['mysql_user'],
                password=self.mysql_path['mysql_password'],
                database=self.mysql_path['mysql_db'],
                connect_timeout=6000,
                charset = 'utf8',
            )

            cursor = mysql_conn.cursor(buffered=True)
            print('mysql_course--->success')
            return mysql_conn, cursor
        except Exception as e:
            logger.info('连接mysql数据库失败-->{}'.format(e))
            return '',''

    def search(self, sql,data):
        """
        查询
        """
        try:
            self.cursor.execute(sql,(data,))
            data = self.cursor.fetchall()

            return data, True
        except Exception as e:
            print('mysql查询错误',e)
            print(sql)
            time.sleep(5)
            self.mysql_conn, self.cursor = self.link_mysql()
            return [], False

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