TEST/common/database_util.py

import traceback
from sqlite3 import Connection, Cursor

# import MySQLdb
import pymysql

from common.log_util import error_log


class DatabaseUtil:
    def __init__(self):
        self.cs = None
        self.conn = None

    def create_conn(self):
        self.conn: Connection = pymysql.connect(
            host='10.12.43.2',
            user='root',
            password='Xx98$kdiesJS',
            database='jwtest',
            port=3409
        )
        return self.conn

    # 执行sql语句
    def execute_sql(self, sql):

        # 创建游标
        self.cs: Cursor = self.create_conn().cursor()
        # 通过游标对象，执行SQL
        try:
            self.cs.execute(sql)
        except:
            error_log("SQL查询异常:%s" % str(traceback.format_exc()))

            # 提取值
        value = self.cs.fetchall()
        self.close_resuoe()
        print(value)
        return value

    def close_resuoe(self):
        self.cs.close()
        self.conn.close()

# if __name__ == '__main__':
#     print(DatabaseUtil().execute_sql('select * from uisql '))